|
50 | 50 | {:k :entity :id id}) |
51 | 51 | (for [[cid mm] (:claims m) :when (>= cid since)] |
52 | 52 | {:k :claim :cid cid :l (:l mm) :p (:p mm) :r (:r mm) :tx (get (:tx-of m) cid)}) |
53 | | - [{:k :tx :tx txid :seq (get-in m [:txs txid :seq]) :agent (get-in m [:txs txid :agent])} |
| 53 | + [{:k :tx :tx txid :seq (get-in m [:txs txid :seq]) :agent (get-in m [:txs txid :agent]) |
| 54 | + :observed (get-in m [:txs txid :observed])} ; causality (thread H): the global seq the writer had SEEN when it decided |
54 | 55 | {:k :commit :tx txid}]))) |
55 | 56 |
|
56 | 57 | ;; --- reads over the reified store ------------------------------------------- |
|
83 | 84 | (defn agent-of [co cid] |
84 | 85 | (let [m @(store co)] (get-in m [:txs (get (:tx-of m) cid) :agent]))) |
85 | 86 |
|
| 87 | +;; --- causality / as-of (thread H, Part A): the causal stamp ------------------ |
| 88 | +;; Every coordination write already reports :base = "the version I had observed when |
| 89 | +;; I decided" (the daemon/CLI path passes the GLOBAL :version it round-tripped; commit!/ |
| 90 | +;; retract! used it ONLY for the single-valued staleness reject, then dropped it). We now |
| 91 | +;; THREAD that base into the tx record as :observed — one int per tx, recovered through |
| 92 | +;; replay exactly like :seq/:agent. This turns happens-before into a recorded fact: |
| 93 | +;; "did peer B's claim exist in the view A read before A acted?" == (<= seq(B) observed(A)). |
| 94 | +;; observed-of reads it; nil for legacy/non-causal writes -> callers fall back to seq-of |
| 95 | +;; (commit order), so the causal election degrades to cid-order, never throws. |
| 96 | +;; RISK GUARD: the writer cannot claim to have observed the FUTURE — observed is clamped |
| 97 | +;; to the pre-commit current-seq at the write site (a backdated stamp only LOSES elections). |
| 98 | +(defn observed-of [co cid] |
| 99 | + (let [m @(store co)] (get-in m [:txs (get (:tx-of m) cid) :observed]))) |
| 100 | +;; the causal key of a live claim: [observed-or-seq, cid, agent]. observed orders by |
| 101 | +;; DECISION time (who saw the empty group first), cid/agent keep it a total order. A LATER |
| 102 | +;; commit (higher cid) that DECIDED earlier (lower observed) wins — this is the whole point: |
| 103 | +;; election by causal view, not by commit order. Pure fn of recorded claims -> every reader |
| 104 | +;; computes it identically with zero coordination. |
| 105 | +(defn causal-key [co cid] |
| 106 | + [(or (observed-of co cid) (seq-of co cid)) cid (str (agent-of co cid))]) |
| 107 | + |
86 | 108 | ;; --- views-as-claims (thread E): per-branch isolation over the same log ------ |
87 | 109 | ;; A VIEW is a first-class subject; (view selects @cid) claims are its OVERLAY — |
88 | 110 | ;; the cids it treats as facts. The object IS a claim id: cids live in the same |
|
119 | 141 | pool (if (seq in-view) in-view cids)] |
120 | 142 | (first (sort-by (fn [cid] [cid (str (agent-of co cid))]) pool)))))) |
121 | 143 |
|
| 144 | +;; elect-causal — the CAUSAL election policy (thread H, Part C): same view-relative |
| 145 | +;; pool as `elect`, but ordered by the CAUSAL key [observed, cid, agent] instead of |
| 146 | +;; [cid, agent]. So of a contended live (l,p) group, the winner is the member whose |
| 147 | +;; writer DECIDED earliest (saw the empty/oldest group), tie-broken by commit order |
| 148 | +;; then agent. This is what lets rival drivers/roles COEXIST and resolve by "who had |
| 149 | +;; the earlier causal view" rather than "who happened to commit first" — both readers |
| 150 | +;; agree, nothing blocks. Degrades to `elect` when no :observed stamps exist (legacy |
| 151 | +;; claims fall back to seq-of via causal-key), so it is a strict refinement, never a |
| 152 | +;; regression. nil on an empty group. |
| 153 | +(defn elect-causal |
| 154 | + ([co cids] (elect-causal co nil cids)) |
| 155 | + ([co view cids] |
| 156 | + (when (seq cids) |
| 157 | + (let [sel (when view (view-selects co view)) |
| 158 | + in-view (when (seq sel) (filterv sel cids)) |
| 159 | + pool (if (seq in-view) in-view cids)] |
| 160 | + (first (sort-by (fn [cid] (causal-key co cid)) pool)))))) |
| 161 | + |
122 | 162 | (defn- ent! [co tx nm] |
123 | 163 | (or (s/resolve-name (store co) nm) |
124 | 164 | (let [e (c/entity! (store co))] (s/name! (store co) e nm tx) e))) |
|
210 | 250 | {:ok (current-seq co) :idempotent true} |
211 | 251 |
|
212 | 252 | :else |
213 | | - (let [since (:next-id @(store co)) |
| 253 | + (let [since (:next-id @(store co)) |
| 254 | + observed (let [pre (current-seq co)] (min (or base pre) pre)) ; causal stamp, clamped to head (no future) |
214 | 255 | tx (c/begin-tx! (store co) agent) |
| 256 | + _ (swap! (store co) assoc-in [:txs tx :observed] observed) |
215 | 257 | te (ent! co tx te-name)] |
216 | 258 | (case kind |
217 | 259 | :link (s/link! (store co) te pred (ent! co tx r-spec) tx) |
|
282 | 324 | (if (empty? victims) |
283 | 325 | {:ok (current-seq co)} |
284 | 326 | (let [since (:next-id @(store co)) |
| 327 | + observed (let [pre (current-seq co)] (min (or base pre) pre)) ; causal stamp on the retract tx |
285 | 328 | tx (c/begin-tx! (store co) agent) |
| 329 | + _ (swap! (store co) assoc-in [:txs tx :observed] observed) |
286 | 330 | sup (c/value! (store co) "cnf-supersedes")] |
287 | 331 | (doseq [old victims] (c/claim! (store co) old sup old tx)) |
288 | 332 | (append-tx! co (delta-records co since tx)) |
|
392 | 436 | ents (vec (for [r recs :when (= (:k r) :entity)] (:id r))) |
393 | 437 | claims (vec (for [r recs :when (= (:k r) :claim)] [(:cid r) {:l (:l r) :p (:p r) :r (:r r)}])) |
394 | 438 | tx-of (vec (for [r recs :when (= (:k r) :claim)] [(:cid r) (:tx r)])) |
395 | | - txs (vec (for [r recs :when (= (:k r) :tx)] [(:tx r) {:seq (:seq r) :agent (:agent r)}])) |
| 439 | + txs (vec (for [r recs :when (= (:k r) :tx)] [(:tx r) {:seq (:seq r) :agent (:agent r) :observed (:observed r)}])) ; recover the causal stamp through replay (acceptance d) |
396 | 440 | sup (some (fn [[id v]] (when (= v "cnf-supersedes") id)) vals) |
397 | 441 | superd (vec (for [[_ m] claims :when (= (:p m) sup)] (:r m))) |
398 | 442 | all-id (concat (map first vals) ents (map first claims) (map first txs)) |
|
425 | 469 | (emit {:k :entity :id id})) |
426 | 470 | (doseq [[cid cl] (:claims m)] |
427 | 471 | (emit {:k :claim :cid cid :l (:l cl) :p (:p cl) :r (:r cl) :tx (get (:tx-of m) cid)})) |
428 | | - (doseq [[tx t] (:txs m)] (emit {:k :tx :tx tx :seq (:seq t) :agent (:agent t)})) |
| 472 | + (doseq [[tx t] (:txs m)] (emit {:k :tx :tx tx :seq (:seq t) :agent (:agent t) :observed (:observed t)})) |
429 | 473 | (emit {:k :commit :tx :migration})) |
430 | 474 | (.force (.getChannel os) true)))) |
431 | 475 |
|
|
0 commit comments