Skip to content

dizzy.engine.dagstore

dizzy.engine.dagstore

dagstore — the content-addressed event DAG the event store is built on.

Cryptographic event ids, parent pointers for order, heads as checkpoint, canonical topological replay, git-style anti-entropy sync. Stdlib only: this subpackage adds nothing to DIZZY's dependency footprint, which is why it can sit in core alongside the shells rather than behind an extra.

It knows nothing about DIZZY — no feat file, no generated classes, not even pydantic. It stores (type: str, payload: dict) and returns hashed Event records. :mod:dizzy.engine.store is the layer that gives those payloads their DIZZY meaning.

NotCanonicalizable

Bases: ValueError

The value falls outside the hashed subset (float, big int, non-str key…).

Source code in dizzy/src/dizzy/engine/dagstore/canonical.py 
33
34
class NotCanonicalizable(ValueError):
    """The value falls outside the hashed subset (float, big int, non-str key…)."""

TamperedEvent

Bases: ValueError

An event's id does not match its content.

Source code in dizzy/src/dizzy/engine/dagstore/events.py 
23
24
class TamperedEvent(ValueError):
    """An event's id does not match its content."""

DagStore

A single node's event DAG. :memory: (default) or a file path.

Source code in dizzy/src/dizzy/engine/dagstore/store.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
class DagStore:
    """A single node's event DAG. ``:memory:`` (default) or a file path."""

    def __init__(self, path: str = ":memory:", check_same_thread: bool = True):
        # check_same_thread=False lets a host share one store across threads;
        # the host must then serialize calls itself (sqlite3 connections are
        # not internally thread-safe).
        self._db = sqlite3.connect(path, check_same_thread=check_same_thread)
        self._db.executescript(_SCHEMA)
        self._db.commit()

    # ── write ────────────────────────────────────────────────────────────

    def append(self, type: str, payload: dict[str, Any]) -> Event:
        """Mint a new event on top of the current heads and store it."""
        with self._db:
            self._db.execute("BEGIN IMMEDIATE")
            event = make_event(type, self.heads(), payload)
            self._insert(event)
        return event

    def add(self, event: Event) -> bool:
        """Ingest a replicated event. Returns False if already present.

        Verifies the content hash (raises TamperedEvent) and requires every
        parent to be present already (raises MissingParents) — replication
        must deliver ancestry first, which `sync` guarantees.
        """
        if not verify(event):
            raise TamperedEvent(f"id {event.id} does not match content")
        with self._db:
            self._db.execute("BEGIN IMMEDIATE")
            if event.id in self:
                return False
            missing = [p for p in event.parents if p not in self]
            if missing:
                raise MissingParents(missing)
            self._insert(event)
        return True

    def _insert(self, event: Event) -> None:
        self._db.execute(
            "INSERT INTO events (id, type, payload) VALUES (?, ?, ?)",
            (event.id, event.type, json.dumps(event.payload, ensure_ascii=False)),
        )
        self._db.executemany(
            "INSERT INTO edges (child, parent) VALUES (?, ?)",
            [(event.id, p) for p in event.parents],
        )

    # ── read ─────────────────────────────────────────────────────────────

    def heads(self) -> tuple[str, ...]:
        """The DAG frontier (sorted): events no known event names as parent."""
        rows = self._db.execute(
            "SELECT id FROM events WHERE id NOT IN (SELECT parent FROM edges) ORDER BY id"
        ).fetchall()
        return tuple(r[0] for r in rows)

    def get(self, event_id: str) -> Event:
        row = self._db.execute(
            "SELECT id, type, payload FROM events WHERE id = ?", (event_id,)
        ).fetchone()
        if row is None:
            raise KeyError(event_id)
        parents = tuple(
            r[0]
            for r in self._db.execute(
                "SELECT parent FROM edges WHERE child = ? ORDER BY parent", (event_id,)
            )
        )
        return Event(id=row[0], type=row[1], parents=parents, payload=json.loads(row[2]))

    def __contains__(self, event_id: str) -> bool:
        return (
            self._db.execute("SELECT 1 FROM events WHERE id = ?", (event_id,)).fetchone()
            is not None
        )

    def __len__(self) -> int:
        return self._db.execute("SELECT count(*) FROM events").fetchone()[0]

    def ids(self) -> frozenset[str]:
        return frozenset(r[0] for r in self._db.execute("SELECT id FROM events"))

    def iterate(self) -> Iterator[Event]:
        """Yield every event in canonical replay order.

        Kahn's algorithm with the ready set kept as a sorted frontier: an
        event becomes ready once all its parents have been emitted; among
        ready events the smallest id goes first. Purely a function of the
        event set — no clocks, no insertion order.
        """
        import heapq

        pending: dict[str, int] = {}  # id -> unemitted parent count
        children: dict[str, list[str]] = {}
        for child, parent in self._db.execute("SELECT child, parent FROM edges"):
            pending[child] = pending.get(child, 0) + 1
            children.setdefault(parent, []).append(child)

        ready = [r[0] for r in self._db.execute("SELECT id FROM events") if r[0] not in pending]
        heapq.heapify(ready)
        emitted = 0
        while ready:
            event_id = heapq.heappop(ready)
            yield self.get(event_id)
            emitted += 1
            for child in children.get(event_id, ()):
                pending[child] -= 1
                if pending[child] == 0:
                    del pending[child]
                    heapq.heappush(ready, child)
        if pending:
            raise RuntimeError(f"DAG has {len(pending)} events with unresolvable parents")

append(type, payload)

Mint a new event on top of the current heads and store it.

Source code in dizzy/src/dizzy/engine/dagstore/store.py 
64
65
66
67
68
69
70
def append(self, type: str, payload: dict[str, Any]) -> Event:
    """Mint a new event on top of the current heads and store it."""
    with self._db:
        self._db.execute("BEGIN IMMEDIATE")
        event = make_event(type, self.heads(), payload)
        self._insert(event)
    return event

add(event)

Ingest a replicated event. Returns False if already present.

Verifies the content hash (raises TamperedEvent) and requires every parent to be present already (raises MissingParents) — replication must deliver ancestry first, which sync guarantees.

Source code in dizzy/src/dizzy/engine/dagstore/store.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def add(self, event: Event) -> bool:
    """Ingest a replicated event. Returns False if already present.

    Verifies the content hash (raises TamperedEvent) and requires every
    parent to be present already (raises MissingParents) — replication
    must deliver ancestry first, which `sync` guarantees.
    """
    if not verify(event):
        raise TamperedEvent(f"id {event.id} does not match content")
    with self._db:
        self._db.execute("BEGIN IMMEDIATE")
        if event.id in self:
            return False
        missing = [p for p in event.parents if p not in self]
        if missing:
            raise MissingParents(missing)
        self._insert(event)
    return True

heads()

The DAG frontier (sorted): events no known event names as parent.

Source code in dizzy/src/dizzy/engine/dagstore/store.py 
103
104
105
106
107
108
def heads(self) -> tuple[str, ...]:
    """The DAG frontier (sorted): events no known event names as parent."""
    rows = self._db.execute(
        "SELECT id FROM events WHERE id NOT IN (SELECT parent FROM edges) ORDER BY id"
    ).fetchall()
    return tuple(r[0] for r in rows)

iterate()

Yield every event in canonical replay order.

Kahn's algorithm with the ready set kept as a sorted frontier: an event becomes ready once all its parents have been emitted; among ready events the smallest id goes first. Purely a function of the event set — no clocks, no insertion order.

Source code in dizzy/src/dizzy/engine/dagstore/store.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def iterate(self) -> Iterator[Event]:
    """Yield every event in canonical replay order.

    Kahn's algorithm with the ready set kept as a sorted frontier: an
    event becomes ready once all its parents have been emitted; among
    ready events the smallest id goes first. Purely a function of the
    event set — no clocks, no insertion order.
    """
    import heapq

    pending: dict[str, int] = {}  # id -> unemitted parent count
    children: dict[str, list[str]] = {}
    for child, parent in self._db.execute("SELECT child, parent FROM edges"):
        pending[child] = pending.get(child, 0) + 1
        children.setdefault(parent, []).append(child)

    ready = [r[0] for r in self._db.execute("SELECT id FROM events") if r[0] not in pending]
    heapq.heapify(ready)
    emitted = 0
    while ready:
        event_id = heapq.heappop(ready)
        yield self.get(event_id)
        emitted += 1
        for child in children.get(event_id, ()):
            pending[child] -= 1
            if pending[child] == 0:
                del pending[child]
                heapq.heappush(ready, child)
    if pending:
        raise RuntimeError(f"DAG has {len(pending)} events with unresolvable parents")

MissingParents

Bases: KeyError

add() was given an event whose parents are not yet in the store.

Source code in dizzy/src/dizzy/engine/dagstore/store.py 
32
33
class MissingParents(KeyError):
    """add() was given an event whose parents are not yet in the store."""

canonical_json(value)

Serialize value to canonical JSON bytes (UTF-8).

Deterministic: equal values (regardless of dict insertion order) always produce identical bytes. Raises NotCanonicalizable for anything outside the hashed subset.

Source code in dizzy/src/dizzy/engine/dagstore/canonical.py 
89
90
91
92
93
94
95
96
97
98
def canonical_json(value: Any) -> bytes:
    """Serialize *value* to canonical JSON bytes (UTF-8).

    Deterministic: equal values (regardless of dict insertion order) always
    produce identical bytes. Raises NotCanonicalizable for anything outside
    the hashed subset.
    """
    out: list[str] = []
    _serialize(value, out)
    return "".join(out).encode("utf-8")

verify(event)

True iff the event's id matches its content.

Source code in dizzy/src/dizzy/engine/dagstore/events.py 
50
51
52
def verify(event: Event) -> bool:
    """True iff the event's id matches its content."""
    return event.id == compute_id(event.type, event.parents, event.payload)