SWE / E-commerce Shipping · AI Assisted Coding

TikTok AI Coding:OrderValidator 动态规则 DAG,删除后怎样继承依赖?

用版本化规则图、拓扑执行与前置失败阻断,深入证明删除中间规则后的边连接不会丢失依赖关系。

题目重点:用版本化规则图、拓扑执行与前置失败阻断,深入证明删除中间规则后的边连接不会丢失依赖关系。

题意与假设

这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。

  • 规则为无副作用bool predicate,边u→v表示v必须等待u且u通过;失败/异常会阻断后继,这是明确的训练执行政策。
  • 删除只取消该规则本身;保留其前置与后置之间的依赖。规则ID唯一,单线程示例,不把多线程snapshot说成自动原子。
  • 示例add_rule只允许新增节点依赖已有节点,不支持偷偷改已有rule函数/加入任意边;实际编辑接口需要另定义版本与并发合同。

解题路线

  1. 把问题分为规则定义、依赖图、执行状态。规则失败不应被表示为“图有环”,异常也与订单不满足价格限制不同;输出passed/failed/error/blocked能解释为何某项没有运行。
  2. 构造时检查所有边端点存在并作拓扑验证。版本更新先在新容器中验证再赋值,错误更新不留下半张图;每次有效编辑递增graph version,审计能关联订单与所用规则集合。
  3. 拓扑执行保证所有前置状态已确定。若任一前置不passed,当前blocked且不调用predicate;否则实际运行,bool false为failed,异常为error。可根据业务改为收集全部错误,但不可和阻断合同混用。
  4. 删除r前取P=parents(r)、C=children(r),删掉r及所有相邻边,再加入P×C。若已有边则set去重,避免拓扑入度被重复边污染。一个前置有多个后置时必须连笛卡尔积,不只挑一条链。
  5. 证明依赖可达性保留:旧路径不经r则仍存在,经u→r→v的段可替换成u→v;新路径用到新增边则展开回旧u→r→v。因此剩余节点间可达性相同。若新图出现环,展开就得到旧图环,与旧DAG矛盾。
  6. 这保留的是依赖关系,不能保证校验结果不变,因为被删predicate不再执行;原先r失败会阻断后继,删r后其前置通过则后继可以运行。这正是删除规则政策,不是测试bug。
  7. 在线每单使用固定graph版本,不在执行中一边跑一边读取被修改的parents。并发snapshot和变更要锁或immutable pointer切换;predicate读取外部数据也需版本/超时。副作用规则不属于当前纯校验模型,不能简单重试。
  8. AI辅助时先写小图和预期状态再接受代码。重点验菱形共享前置、节点多前置/多后置、重复边、非法新依赖、失败阻断、不执行blocked规则;独立验证比较移除前后的传递闭包,不仅看拓扑可运行。

手动走一遍

A→R、B→R、R→C、R→D,删除R后应有A→C/A→D/B→C/B→D。若A通过、B失败,C/D均blocked。若只连接A→C和B→D,C可能在B失败时错误通过。图A→R→C删除R后保留A→C,但不再执行R的价格检查。

Python 参考实现

下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。

代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。

"""Pure validation rules in a versioned DAG. Edges prerequisite -> dependent.
Remove a node by connecting each former prerequisite to each former dependent.
False/exception blocks descendants; evaluate on a graph snapshot per request.
"""
from collections import deque

class RuleGraph:
    def __init__(self, rules, edges):
        self.rules = dict(rules)
        self.edges = set(edges)
        self.version = 0
        self._order(self.rules, self.edges)

    @staticmethod
    def _order(rules, edges):
        children = {key: [] for key in rules}
        indegree = dict.fromkeys(rules, 0)
        for u, v in edges:
            if u not in rules or v not in rules:
                raise ValueError("unknown rule")
            children[u].append(v)
            indegree[v] += 1
        ready = deque(key for key in rules if indegree[key] == 0)
        order = []
        while ready:
            u = ready.popleft()
            order.append(u)
            for v in children[u]:
                indegree[v] -= 1
                if indegree[v] == 0:
                    ready.append(v)
        if len(order) != len(rules):
            raise ValueError("cycle")
        return order

    def add_rule(self, key, predicate, prerequisites=()):
        if key in self.rules or not callable(predicate):
            raise ValueError("new callable rule required")
        rules = dict(self.rules)
        rules[key] = predicate
        edges = self.edges | {(u, key) for u in prerequisites}
        self._order(rules, edges)  # validate before committing
        self.rules, self.edges = rules, edges
        self.version += 1

    def remove_rule(self, key):
        if key not in self.rules:
            raise ValueError("unknown rule")
        parents = {u for u, v in self.edges if v == key}
        children = {v for u, v in self.edges if u == key}
        rules = {u: f for u, f in self.rules.items() if u != key}
        edges = {(u, v) for u, v in self.edges if u != key and v != key}
        edges |= {(u, v) for u in parents for v in children}
        self._order(rules, edges)
        self.rules, self.edges = rules, edges
        self.version += 1

    def validate(self, order):
        # Snapshot containers. In concurrent use, snapshot and mutations require a lock.
        rules, edges, version = dict(self.rules), set(self.edges), self.version
        parents = {u: [] for u in rules}
        for u, v in edges:
            parents[v].append(u)
        statuses = {}
        for key in self._order(rules, edges):
            if any(statuses[p] != "passed" for p in parents[key]):
                statuses[key] = "blocked"
                continue
            try:
                result = rules[key](order)
                if not isinstance(result, bool):
                    raise ValueError("predicate must return bool")
                statuses[key] = "passed" if result else "failed"
            except Exception:
                statuses[key] = "error"
        return version, statuses

复杂度

拓扑验证/执行O(V+E)加所有实际predicate成本。删除一个节点新增至多|P|·|C|条边,可达O(V+E+|P|·|C|)时间/空间;多次删除可造成边密度上升,不能说每次O(1)。

易错点与追问

容易失分的边界

  • 只连一对前置/后置。
  • 新增重复边但入度重复累加。
  • 删rule后仍要求所有校验结果不变。
  • blocked规则仍执行有副作用。
  • 并发变更期间读取混合版本。

继续深挖

  • 若依赖只要求先运行而非通过怎么改?
  • 边爆炸时怎样以虚拟节点保留依赖?
  • 一次订单使用哪个规则版本?
  • 规则执行超时和外部数据变化怎样审计?
✓已复制微信号!