Design a rule and workflow engine

01The design, in the six-step framework

1. Requirements and use cases

A rule engine separates what the business decides from the code that runs it. An old Google onsite asked a compact version: given a table of allow and deny rules on source and target IP and port, decide whether a request passes. The interviewer called it network object-oriented design. Generalised, that gives this shape.

• An operator defines rules as a condition and an action, with a priority, from a configuration document.
• A caller evaluates a fact (a request, an order) and gets the actions that fired.
• The engine supports two modes: first match wins, or all matches fire.
• A developer registers new condition and action types at runtime without changing the engine.
• An operator sees which rule fired and why.
• A workflow chains steps: a rule's action may update the fact for the next rule.

Out of scope: forward-chaining inference such as Rete, scheduling, persistence and a rule editor. Non-functional constraints: configuration errors are reported at load time, never during evaluation; evaluation is deterministic given priority order; the engine has no knowledge of any business domain.

2. Core entities and relationships

• Fact: a dictionary of attributes being judged.
• Condition: an interface with matches(fact); leaf comparisons and And, Or, Not.
• Action: an interface with apply(fact).
• Rule: name, priority, condition, action.
• Registry: maps a type name in the configuration to a constructor.
• RuleLoader: turns a configuration document into rules, validating as it goes.
• RuleEngine: runs ordered rules in a chosen mode and returns a trace.

A rule composes its condition and action, which are built for it from configuration. The engine composes its ordered rules. The registries belong to the loader, so evaluation has no dependency on how rules were created.

3. Interfaces and key methods

class Condition(Protocol):
    def matches(self, fact: dict) -> bool: ...

class Action(Protocol):
    def apply(self, fact: dict) -> None: ...

class Registry:
    def register(self, name: str, ctor: Callable[..., Any]) -> None:
        """Refuses a duplicate name."""
    def create(self, spec: dict) -> Any:
        """Build from {"type": name, ...}; raises RuleConfigError on an unknown type."""

class RuleLoader:
    def load(self, config: list[dict]) -> list[Rule]:
        """Validate everything first; a bad rule fails the whole load."""

class RuleEngine:
    def evaluate(self, fact: dict) -> Result:
        """Rules in descending priority; ties by definition order; stops at first match in FIRST mode."""

Invariants: the rule list is sorted before the first evaluation and not changed during one; an engine instance is immutable after construction, and reload builds a new one; a failing action is reported in the trace, never swallowed silently.

4. Design patterns and the choices behind them

Conditions as specifications

A rule's test is a tree of small predicates. The specification pattern gives each predicate a class and combinators for and, or and not, so "source in 10.0.0.0/8 and port is 22" is data in a document. The firewall version is exactly this with CidrMatch and Equals.

Evaluation as a chain

Rules are tried in order until one handles the fact: a chain of responsibility in FIRST mode, and a plain pipeline in ALL mode. Making the mode a parameter means the firewall (first match) and a discount engine (all matches) share the code.

A registry for plugins

A registry maps a type name to a constructor. New behaviour is registered from outside, so the engine satisfies the open-closed principle literally: no edit to add a GeoMatch. Keeping Condition and Action single-method is interface segregation, so plugin authors implement one thing. A chain of if type == ... in the loader is the rejected form.

Immutable engine, swapped on reload

Rule changes arrive while evaluations run. Building a new engine and swapping one reference means no evaluation sees half a rule set, with no lock on the hot path.

5. Core flow and code

1. RuleLoader.load walks the configuration, building each condition recursively through the registry.
2. RuleEngine sorts by descending priority, stable on definition order.
3. evaluate tests each rule, recording a trace line.
4. A match applies the action; a failing action is recorded; FIRST mode stops.
5. The firewall builds a fact from a request and reads verdict, which defaults to deny.

from __future__ import annotations
import ipaddress
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Protocol


class RuleConfigError(Exception): ...


class Condition(Protocol):
    def matches(self, fact: dict) -> bool: ...


class Action(Protocol):
    def apply(self, fact: dict) -> None: ...


@dataclass(frozen=True)
class Equals:
    field_name: str
    value: Any
    def matches(self, fact: dict) -> bool:
        return fact.get(self.field_name) == self.value


@dataclass(frozen=True)
class CidrMatch:
    field_name: str
    network: ipaddress.IPv4Network
    def matches(self, fact: dict) -> bool:
        return ipaddress.ip_address(fact[self.field_name]) in self.network


@dataclass(frozen=True)
class AllOf:
    parts: tuple[Condition, ...]
    def matches(self, fact: dict) -> bool:
        return all(p.matches(fact) for p in self.parts)


@dataclass(frozen=True)
class AnyOf:
    parts: tuple[Condition, ...]
    def matches(self, fact: dict) -> bool:
        return any(p.matches(fact) for p in self.parts)


@dataclass(frozen=True)
class Not:
    inner: Condition
    def matches(self, fact: dict) -> bool:
        return not self.inner.matches(fact)


@dataclass(frozen=True)
class SetField:
    field_name: str
    value: Any
    def apply(self, fact: dict) -> None:
        fact[self.field_name] = self.value


@dataclass(frozen=True)
class Rule:
    name: str
    priority: int
    condition: Condition
    action: Action


class Registry:
    def __init__(self) -> None:
        self.ctors: dict[str, Callable[..., Any]] = {}

    def register(self, name: str, ctor: Callable[..., Any]) -> None:
        if name in self.ctors:
            raise RuleConfigError(f"duplicate type {name}")
        self.ctors[name] = ctor

    def create(self, spec: dict) -> Any:
        kind = spec.get("type")
        if kind not in self.ctors:
            raise RuleConfigError(f"unknown type {kind}")
        return self.ctors[kind](spec)


class RuleLoader:
    def __init__(self, conditions: Registry, actions: Registry) -> None:
        self.conditions, self.actions = conditions, actions

    def load(self, config: list[dict]) -> list[Rule]:
        return [Rule(c["name"], c.get("priority", 0), self.conditions.create(c["when"]),
                     self.actions.create(c["then"])) for c in config]


def default_registries() -> tuple[Registry, Registry]:
    cond, act = Registry(), Registry()
    cond.register("equals", lambda s: Equals(s["field"], s["value"]))
    cond.register("cidr", lambda s: CidrMatch(s["field"], ipaddress.ip_network(s["network"])))
    cond.register("all", lambda s: AllOf(tuple(cond.create(p) for p in s["of"])))
    cond.register("any", lambda s: AnyOf(tuple(cond.create(p) for p in s["of"])))
    cond.register("not", lambda s: Not(cond.create(s["of"])))
    act.register("set", lambda s: SetField(s["field"], s["value"]))
    return cond, act


class Mode(Enum):
    FIRST = "first"
    ALL = "all"


@dataclass
class Result:
    fired: list[str] = field(default_factory=list)
    trace: list[str] = field(default_factory=list)


class RuleEngine:
    def __init__(self, rules: list[Rule], mode: Mode = Mode.FIRST) -> None:
        self.rules = tuple(sorted(rules, key=lambda r: -r.priority))
        self.mode = mode

    def evaluate(self, fact: dict) -> Result:
        result = Result()
        for rule in self.rules:
            if not rule.condition.matches(fact):
                result.trace.append(f"{rule.name}: no match")
                continue
            try:
                rule.action.apply(fact)
                result.fired.append(rule.name)
                result.trace.append(f"{rule.name}: fired")
            except Exception as e:
                result.trace.append(f"{rule.name}: action failed: {e}")
            if self.mode is Mode.FIRST:
                break
        return result


class Firewall:
    """The reported interview question, as one small client of the engine."""
    def __init__(self, engine: RuleEngine) -> None:
        self.engine = engine

    def allows(self, src: str, dst: str, port: int) -> bool:
        fact = {"src": src, "dst": dst, "port": port, "verdict": "deny"}
        self.engine.evaluate(fact)
        return fact["verdict"] == "allow"

6. Extensions, concurrency and what interviewers probe

• Thousands of firewall rules: a linear scan per packet is too slow. Index rules by port and by a prefix trie on the source address, so that only candidates are tested, preserving priority order. The engine interface stays; only the candidate selection changes.
• Conflicting rules: first match with priorities gives one defined answer. Say which mode the domain needs, and that default-deny belongs in the fact, not in a final catch-all rule someone can delete.
• Workflow: a workflow engine is rules plus state: each step's action sets fact["state"], and conditions test it. In ALL mode beware of loops; keep a maximum step count.
• Safety of plugins: a registered action can do anything, so restrict the registry to trusted code; configuration should never evaluate strings as code.
• Testing: because each condition is a value, it can be unit-tested alone, and the trace makes end-to-end assertions readable.
• Test first: a higher-priority deny beats a lower allow; unknown type fails at load; ALL mode fires two rules in priority order; an action that raises appears in the trace and does not stop other rules in ALL mode.
• Common mistake: writing a parser for a rule language first. A JSON tree gets you the whole design in the time you have.

02Where this question was reported

• Codemia · Codemia Hard; also /design-a-workflow-engine and /design-a-business-rule-engine
• 1point3acres · Interviewer called it a network design OOD; given a config table of allow/deny maps, write the function deciding whether a request passes