题目重点:用版本化规则图、拓扑执行与前置失败阻断,深入证明删除中间规则后的边连接不会丢失依赖关系。
题意与假设
这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。
- 规则为无副作用bool predicate,边u→v表示v必须等待u且u通过;失败/异常会阻断后继,这是明确的训练执行政策。
- 删除只取消该规则本身;保留其前置与后置之间的依赖。规则ID唯一,单线程示例,不把多线程snapshot说成自动原子。
- 示例add_rule只允许新增节点依赖已有节点,不支持偷偷改已有rule函数/加入任意边;实际编辑接口需要另定义版本与并发合同。
解题路线
- 把问题分为规则定义、依赖图、执行状态。规则失败不应被表示为“图有环”,异常也与订单不满足价格限制不同;输出passed/failed/error/blocked能解释为何某项没有运行。
- 构造时检查所有边端点存在并作拓扑验证。版本更新先在新容器中验证再赋值,错误更新不留下半张图;每次有效编辑递增graph version,审计能关联订单与所用规则集合。
- 拓扑执行保证所有前置状态已确定。若任一前置不passed,当前blocked且不调用predicate;否则实际运行,bool false为failed,异常为error。可根据业务改为收集全部错误,但不可和阻断合同混用。
- 删除r前取P=parents(r)、C=children(r),删掉r及所有相邻边,再加入P×C。若已有边则set去重,避免拓扑入度被重复边污染。一个前置有多个后置时必须连笛卡尔积,不只挑一条链。
- 证明依赖可达性保留:旧路径不经r则仍存在,经u→r→v的段可替换成u→v;新路径用到新增边则展开回旧u→r→v。因此剩余节点间可达性相同。若新图出现环,展开就得到旧图环,与旧DAG矛盾。
- 这保留的是依赖关系,不能保证校验结果不变,因为被删predicate不再执行;原先r失败会阻断后继,删r后其前置通过则后继可以运行。这正是删除规则政策,不是测试bug。
- 在线每单使用固定graph版本,不在执行中一边跑一边读取被修改的parents。并发snapshot和变更要锁或immutable pointer切换;predicate读取外部数据也需版本/超时。副作用规则不属于当前纯校验模型,不能简单重试。
- 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规则仍执行有副作用。
- 并发变更期间读取混合版本。
继续深挖
- 若依赖只要求先运行而非通过怎么改?
- 边爆炸时怎样以虚拟节点保留依赖?
- 一次订单使用哪个规则版本?
- 规则执行超时和外部数据变化怎样审计?
