OpenAgents

Commit

Tell a node that is behind the log apart from one that disagrees with it

6bef9222b8cc · AtlantisPleb · · parent 992ffa079c93 

Tell a node that is behind the log apart from one that disagrees with it

The WAL is shared and the projections are not. Every node replays entries
when it reads, so for the minute after a push lands elsewhere a node is
legitimately a few entries behind — and the verifier reported that as
`served_refs_diverged` plus `object_missing`, the two findings that mean the
served state contradicts the record. On a healthy forge under ordinary push
traffic they fired on some repository most of the time, on whichever node
answered, which is how the check that would catch a real one becomes the
check an operator learns to skip. #179's scheduled pass would have shipped
pre-tuned-out.

Verification now locates the projection on the log instead of comparing it to
the head. The greatest sequence whose recorded post-state refs the repository
serves exactly is its position; `head_seq - position` is how far behind it is.
A projection sitting at some sequence carries no finding. One sitting at no
sequence still reports every ref that differs, because lag runs one way only:
a node that has not replayed an entry is missing what that entry introduced
and can never serve a ref the log has no record of, a ref at a value the log
never recorded, or a value recorded before the state it serves. `object_missing`
is bounded by the position rather than waived.

The applied-sequence marker bounds the search rather than answering it. Only
sequences at or above it are candidates, so rolling it forward makes the check
stricter, not quieter, and rolling it back to -1 admits only older states the
log itself records. A marker past the end of the log is `applied_seq_beyond_log`,
which is how a WAL truncated at the tail — contiguous afterward, invisible to
every other check — surfaces on a node that had applied past it.

`verify_cluster/2` is the fleet's answer. Each report names its node, applied
sequence, position, and distance from the head; the combined verdict is
`:converging` while members are behind and `:diverged` only when a member
contradicts the log, cannot see a repository the others verified, or reports a
different chain link at the same sequence. A verdict that flickers because one
node is replaying is not useful, and this one does not.

Two cases are named rather than hidden: a projection rolled back to a state the
log passed through with its marker rolled back to match is reported as behind,
because from the WAL and the repository alone that is the same observation; and
a deleted cache is an empty projection at sequence -1, which is what a node
that has never replayed looks like. The independence test that asserted the
second was tampering now asserts it is behind by the whole log, which is still
not clean.

Closes #251.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnhfrafYx5ZGaMbzZEJQ2d
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Closes
#251

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

pushed
by user · WAL seq 405 · 2026-08-25T15:38:51.383390Z

Changed files

  • modified INVARIANTS.md
  • modified docs/taxonomy.md
  • modified lib/openagents/forge/verification.ex
  • modified test/openagents/forge/independence_test.exs
  • added test/openagents/forge/verification_test.exs

Diff

5 files changed, +910 -41

INVARIANTS.md modified +49 -2

@@ -4612,9 +4612,56 @@ cache. This proves that divergence between the WAL and what is served is

4612 4612 
detectable. The first is about replay; the second is about detection, and
4613 4613 
neither substitutes for the other.
4614 4614
4615 
Amended 2026-08-25 (issue #251). The WAL is shared and the projections are not,
4616 
so a node is routinely a few entries behind a push another node accepted
4617 
seconds ago. Reporting that as `served_refs_diverged` and `object_missing`,
4618 
which is what this did, made the two findings that mean "the served state
4619 
contradicts the record" fire on a healthy forge under ordinary push traffic —
4620 
so the check that would catch a real one is the check an operator learns to
4621 
skip. Lag and divergence are now separated, and the separation is a property of
4622 
the log rather than a tolerance.
4623
4624 
A projection is *located* on the log: the greatest sequence whose recorded
4625 
post-state refs it serves exactly is its position, and `head_seq - position` is
4626 
how far behind it is. A projection at some sequence carries no finding and
4627 
reports `status: :behind`; a projection at no sequence reports
4628 
`served_refs_diverged` against the state it claims, and `status: :diverged`.
4629 
Lag runs one way only, which is what keeps the boundary tight: a node that has
4630 
not replayed an entry is missing what that entry introduced and can never serve
4631 
a ref the log has no record of, a ref at a value the log never recorded, or a
4632 
value recorded before the state it is serving. `object_missing` is bounded by
4633 
the position rather than waived: every object an entry at or below it
4634 
introduced must still be present.
4635
4636 
The applied-sequence marker `OpenAgents.Forge.Sync` writes is the projection's
4637 
own claim, so it bounds the search rather than answering it. Only sequences at
4638 
or above it are candidates, which means rolling the marker forward makes the
4639 
check stricter rather than quieter, and rolling it back to `-1` admits only
4640 
older states the log itself records. A marker naming a sequence the log does
4641 
not have is `applied_seq_beyond_log`, which is how a WAL truncated at the tail
4642 
— contiguous `0..n-1` afterward, and so invisible to every other check here —
4643 
surfaces on a node that had already applied past it.
4644
4645 
Two cases are named rather than hidden. A projection rolled back to a state the
4646 
log passed through, with its marker rolled back to match, is reported as behind:
4647 
from the WAL and the repository alone, that is the same observation as a node
4648 
mid-replay. A deleted cache is an empty projection at sequence `-1`, which is
4649 
what a node that has never replayed looks like; the report still says how far
4650 
from the log it is, and `REPOSITORY-003` is what brings it back.
4651
4652 
Because a node answers only for its own projection, a report names the node, its
4653 
applied sequence, its position, and its distance from the head.
4654 
`verify_cluster/2` combines the members' answers into a verdict lag cannot move:
4655 
`:converging` while nothing contradicts the log and some member is behind or
4656 
silent, and `:diverged` only when a member contradicts the log, when a member
4657 
cannot see a repository the others verified, or when two members report
4658 
different chain links at the same sequence. A scheduled pass (#179) can
4659 
therefore publish findings without publishing the fleet's replay window.
4660
4615 4661 
Evidence: `OpenAgents.Forge.Verification`, `OpenAgents.Forge.RepoRef`,
4616 4662 
`OpenAgents.Forge.WAL`, `OpenAgents.Forge.Repos`,
4617 
`test/openagents/forge/independence_test.exs`, and
4663 
`test/openagents/forge/independence_test.exs`,
4664 
`test/openagents/forge/verification_test.exs`, and
4618 4665 
`test/openagents/forge/repo_ref_test.exs`.
4619 4666
4620 4667 
### EXIT-003 — Recovery comes from the WAL, and the mirror is strictly lossy

@@ -5856,7 +5903,7 @@ contract; the invariant prose above defines the assertion, not the filename.

5856 5903 
| REPOSITORY-002 | `ops/ci/push-remote-check.sh`, `ops/dev/install-push-guard.sh`, `test/openagents/push_remote_contract_test.exs` |
5857 5904 
| REPOSITORY-003 | `test/openagents/forge/wal_replay_test.exs`, `test/openagents/forge/sync_test.exs`, `test/openagents/forge/independence_test.exs` |
5858 5905 
| EXIT-001 | `test/openagents/data_rights/export_inventory_test.exs`, `test/openagents/data_rights/account_export_test.exs` |
5859 
| EXIT-002 | `test/openagents/forge/independence_test.exs` |
5906 
| EXIT-002 | `test/openagents/forge/independence_test.exs`, `test/openagents/forge/verification_test.exs` |
5860 5907 
| EXIT-003 | `test/openagents/forge/independence_test.exs` |
5861 5908 
| EXIT-004 | `test/openagents/forge/independence_test.exs` |
5862 5909 
| EXIT-005 | `test/openagents/forge/independence_test.exs`, `test/openagents/forge/wal_test.exs`, `test/openagents/forge/git_http_test.exs`, `test/openagents_web/controllers/push_receipt_controller_test.exs`, `test/openagents_web/controllers/forge_anchor_controller_test.exs` |
docs/taxonomy.md modified +11

@@ -70,6 +70,17 @@ with `OpenAgents.Forge.RepoRef` before it reaches a path — a name is a legal

70 70 
path segment, so using one as a key silently builds a directory that projects
71 71 
nothing, which is what issue #190 found on the live node.
72 72
73 
**Behind and diverged** — two things a node's bare repository can be, and never
74 
the same word. The repository is a projection of the WAL, each node keeps its
75 
own, and a node **behind** the log has simply not replayed an entry yet: it
76 
serves a state the log records, and it catches up on its own. A node
77 
**diverged** from the log serves a state the log does not record at or after
78 
the sequence that node claims to have applied, and nothing about that clears
79 
itself. `OpenAgents.Forge.Verification.verify/2` reports which, with how far
80 
behind and on which node; `verify_cluster/2` calls a fleet **converging** while
81 
members are behind and **diverged** only when one contradicts the log. Never
82 
write "diverged" for a node that is merely behind (EXIT-002, issue #251).
83
73 84 
**MirrorWatch** — the component that exports accepted `main` commits from the
74 85 
forge to GitHub. GitHub is a mirror only; nothing on GitHub can affect what
75 86 
the forge serves.
lib/openagents/forge/verification.ex modified +448 -35

@@ -10,8 +10,7 @@ defmodule OpenAgents.Forge.Verification do

10 10 
  matches the record, and it reaches that answer without PostgreSQL, without an
11 11 
  operator credential, and without any row an operator can edit.
12 12
13 
  Four findings are possible, and each one names a distinct way the two can
14 
  disagree:
13 
  Each finding names a distinct way the two can disagree:
15 14
16 15 
  * `entry_object_missing` — the index names an entry the store cannot produce.
17 16 
  * `entry_digest_mismatch` — the entry the store produced is not the entry the

@@ -20,11 +19,15 @@ defmodule OpenAgents.Forge.Verification do

20 19 
    changes the key it should have and the recorded key stops matching.
21 20 
  * `entry_sequence_broken` — the entries are not the contiguous run `0..n-1`.
22 21 
    A removed or renumbered entry shows up here.
23 
  * `served_refs_diverged` — a ref the repository serves is not the ref the WAL
24 
    last recorded for it, in either direction. A ref moved on disk without a
25 
    push produces this.
22 
  * `served_refs_diverged` — the repository serves a ref state the WAL never
23 
    recorded at or after the sequence this projection claims, in either
24 
    direction. A ref moved on disk without a push produces this; a ref an entry
25 
    this node has not replayed yet does not, and the next section is where that
26 
    line is drawn.
27 
  * `applied_seq_beyond_log` — the projection's applied-sequence marker names a
28 
    sequence the log does not have, so it claims an entry that is not there.
26 29 
  * `object_missing` — the repository cannot produce an object the WAL says a
27 
    push introduced.
30 
    push it has applied introduced.
28 31 
  * `object_unreachable` — the repository produces every ref tip but cannot
29 32 
    produce something an advertised ref reaches, so a clone aborts partway
30 33 
    through the walk even though every tip resolves.

@@ -47,6 +50,69 @@ defmodule OpenAgents.Forge.Verification do

47 50 
  which is what made an absent repository report `wal_unreadable` — the finding
48 51 
  that means "your log is gone" — for a log that was never there (issue #190).
49 52
53 
  ## Behind the log is not the same as disagreeing with it
54
55 
  One WAL, many projections. Every node keeps its own bare repository on its
56 
  own disk and brings it up by replaying entries when it reads
57 
  (`OpenAgents.Forge.Sync`), so a node is routinely a few entries behind a push
58 
  another node accepted seconds ago. That is the ordinary state of a healthy
59 
  fleet, and a verifier that calls it tampering is a verifier people learn to
60 
  ignore — which is what this reported until issue #251.
61
62 
  The two are told apart by locating the projection on the log rather than
63 
  comparing it to the head. The served ref map is compared against the ref map
64 
  each entry recorded as its post-state, and the greatest sequence whose
65 
  recorded state the projection serves exactly is its `:position`.
66
67 
  * A projection that sits at some sequence on the log is **behind** it by
68 
    `head_seq - position` entries. No finding, `status: :behind`, and
69 
    `{:ok, report}` — nothing contradicts the log.
70 
  * A projection that sits at no sequence on the log **disagrees** with it.
71 
    `served_refs_diverged` names each ref, `status: :diverged`, and
72 
    `{:error, report}`.
73
74 
  Lag only ever runs one way, which is what makes the boundary tight. A node
75 
  that has not replayed an entry is missing what that entry introduced; it can
76 
  never serve a ref the log has no record of, a ref at a value the log never
77 
  recorded, or a ref value the log recorded *before* the state the node is
78 
  serving. Each of those is still a finding however far behind the node is, and
79 
  `object_missing` is still a finding for every object an entry at or below the
80 
  projection's position introduced.
81
82 
  The applied-sequence marker `OpenAgents.Forge.Sync` writes
83 
  (`OpenAgents.Forge.Repos.applied_seq/1`) is the projection's own claim about
84 
  itself, so it bounds the search rather than answering it. Only sequences at
85 
  or above the marker are candidates: a projection serving a state older than
86 
  the sequence it claims to have applied is reported, and a marker rolled
87 
  forward to the head therefore silences nothing — it makes the check stricter,
88 
  because the head's refs still have to be there. A marker rolled back to `-1`
89 
  widens the search to the whole log and to the empty state, which is the one
90 
  thing it buys and it buys nothing else: every candidate is still a state the
91 
  log itself records, so no ref value the log never held is ever admitted. A
92 
  marker naming a sequence the log does not have is `applied_seq_beyond_log`,
93 
  which is how a log truncated at the tail surfaces on a node that had already
94 
  applied past it.
95
96 
  What this cannot separate is named rather than hidden. A projection rolled
97 
  back to a state the log itself passed through, with its marker rolled back to
98 
  match, is indistinguishable from a node that has not finished replaying,
99 
  because from the WAL and the repository alone those are the same observation.
100 
  It is reported as behind, not as clean, and a node that stays behind while
101 
  its peers catch up is what `verify_cluster/2` makes visible.
102
103 
  ## One repository, three answers
104
105 
  Because each node answers for its own projection, `verify/2` answers for one
106 
  node and says which: the report carries `:node`, `:applied_seq`,
107 
  `:position`, `:behind`, and `:head_seq`. `verify_cluster/2` asks every member
108 
  and combines the answers into one verdict that does not flicker while a node
109 
  replays: `:verified` when every member answered and every one is current,
110 
  `:converging` when nothing contradicts the log and some member is behind or
111 
  silent, `:diverged` when some member contradicts the log or two members
112 
  disagree about the log itself, and `:unavailable` when nothing was checked.
113 
  Lag moves a fleet between `:verified` and `:converging`, which is a state
114 
  that clears itself; only a real disagreement reaches `:diverged`.
115
50 116 
  What this cannot do is stated as plainly as what it can. Content addressing
51 117 
  and the chain make tampering *evident*, not *impossible*. An operator who
52 118 
  rewrites an entry, its key, the index, and every link after it produces a

@@ -63,9 +129,11 @@ defmodule OpenAgents.Forge.Verification do

63 129 
  only an operator who serves something other than what was pushed.
64 130 
  """
65 131
132 
  alias OpenAgents.Cluster
66 133 
  alias OpenAgents.Forge.{RepoRef, Repos, WAL}
67 134
68 135 
  @internal_ref_prefix "refs/internal/"
136 
  @cluster_timeout_ms 30_000
69 137
70 138 
  @typedoc "One disagreement between the WAL and what the repository serves."
71 139 
  @type finding :: %{code: String.t(), detail: map()}

@@ -73,14 +141,47 @@ defmodule OpenAgents.Forge.Verification do

73 141 
  @typedoc "An independently held commitment to one entry's link."
74 142 
  @type anchor :: %{seq: non_neg_integer(), link: String.t()}
75 143
76 
  @typedoc "The verification outcome for one repository."
144 
  @typedoc "Where one node's projection sits relative to the log."
145 
  @type status :: :current | :behind | :diverged | :unresolved
146
147 
  @typedoc "The verification outcome for one repository, on one node."
77 148 
  @type report :: %{
78 149 
          repo: RepoRef.ref(),
79 150 
          storage_key: RepoRef.storage_key() | nil,
151 
          node: node(),
80 152 
          entries: non_neg_integer(),
81 153 
          findings: [finding()],
82 154 
          head: anchor() | nil,
83 
          chained_from: non_neg_integer() | nil
155 
          chained_from: non_neg_integer() | nil,
156 
          status: status(),
157 
          head_seq: integer(),
158 
          applied_seq: integer(),
159 
          position: integer() | nil,
160 
          behind: non_neg_integer() | nil
161 
        }
162
163 
  @typedoc "One member's contribution to a fleet-wide answer."
164 
  @type node_result :: %{
165 
          node: node(),
166 
          status: status() | :unreachable,
167 
          head_seq: integer() | nil,
168 
          applied_seq: integer() | nil,
169 
          position: integer() | nil,
170 
          behind: non_neg_integer() | nil,
171 
          entries: non_neg_integer() | nil,
172 
          head: anchor() | nil,
173 
          findings: [finding()],
174 
          reason: term() | nil
175 
        }
176
177 
  @typedoc "The verification outcome for one repository, across the fleet."
178 
  @type cluster_report :: %{
179 
          repo: RepoRef.ref(),
180 
          status: :verified | :converging | :diverged | :unavailable,
181 
          head_seq: integer() | nil,
182 
          log_agreement: :agreed | :disagreed | :unknown,
183 
          nodes: [node_result()],
184 
          findings: [%{node: node(), code: String.t(), detail: map()}]
84 185 
        }
85 186
86 187 
  @doc """

@@ -110,12 +211,21 @@ defmodule OpenAgents.Forge.Verification do

110 211 
  what a caller remembers so it can anchor a later verification, and
111 212 
  `:chained_from`, the first sequence that carries a link. Entries before that
112 213 
  sequence predate the chain and are not covered by it.
214
215 
  This answers for one node's projection, and says so: `:node`, `:applied_seq`,
216 
  `:position`, `:behind`, and `:head_seq` place the answer on the log. A node
217 
  that has not replayed an entry yet returns `{:ok, report}` with
218 
  `status: :behind` and no findings, because being behind is not a
219 
  disagreement. Use `verify_cluster/2` for the fleet's answer.
113 220 
  """
114 221 
  @spec verify(RepoRef.ref(), keyword()) :: {:ok, report()} | {:error, report()}
115 222 
  def verify(repo_ref, opts \\ []) when is_binary(repo_ref) and is_list(opts) do
116 223 
    case RepoRef.storage_key(repo_ref) do
117 
      {:ok, storage_key} -> verify_storage_key(repo_ref, storage_key, opts)
118 
      {:error, reason} -> report(repo_ref, nil, [], [resolution_finding(repo_ref, reason)])
224 
      {:ok, storage_key} ->
225 
        verify_storage_key(repo_ref, storage_key, opts)
226
227 
      {:error, reason} ->
228 
        unresolved(repo_ref, nil, [resolution_finding(repo_ref, reason)])
119 229 
    end
120 230 
  end
121 231

@@ -123,20 +233,22 @@ defmodule OpenAgents.Forge.Verification do

123 233 
    case WAL.read_index(storage_key) do
124 234 
      {:ok, _generation, index} ->
125 235 
        entries = WAL.entries(index)
236 
        projection = locate(storage_key, entries)
126 237
127 238 
        findings =
128 239 
          sequence_findings(entries) ++
129 240 
            entry_findings(storage_key, entries) ++
130 
            ref_findings(storage_key, index) ++
131 
            object_findings(storage_key, entries) ++
132 
            reachability_findings(storage_key, index) ++
241 
            marker_findings(projection) ++
242 
            ref_findings(entries, projection) ++
243 
            object_findings(storage_key, entries, projection) ++
244 
            reachability_findings(storage_key, entries, projection) ++
133 245 
            chain_findings(entries) ++
134 246 
            anchor_findings(entries, normalize_anchor(opts[:anchor]))
135 247
136 
        report(repo_ref, storage_key, entries, findings)
248 
        report(repo_ref, storage_key, entries, findings, projection)
137 249
138 250 
      {:error, reason} ->
139 
        report(repo_ref, storage_key, [], [
251 
        unresolved(repo_ref, storage_key, [
140 252 
          finding("wal_unreadable", %{"reason" => inspect(reason)})
141 253 
        ])
142 254 
    end

@@ -151,19 +263,50 @@ defmodule OpenAgents.Forge.Verification do

151 263 
  defp resolution_finding(repo_ref, _not_found),
152 264 
    do: finding("repository_not_found", %{"repo" => repo_ref})
153 265
154 
  defp report(repo, storage_key, entries, findings) do
266 
  # Nothing was compared: the reference named no single repository, or the log
267 
  # could not be read. A node that could not check is not a node that found
268 
  # something, and a fleet answer must not read one as the other.
269 
  defp unresolved(repo, storage_key, findings) do
270 
    {:error,
271 
     %{
272 
       repo: repo,
273 
       storage_key: storage_key,
274 
       node: node(),
275 
       entries: 0,
276 
       findings: findings,
277 
       head: nil,
278 
       chained_from: nil,
279 
       status: :unresolved,
280 
       head_seq: -1,
281 
       applied_seq: -1,
282 
       position: nil,
283 
       behind: nil
284 
     }}
285 
  end
286
287 
  defp report(repo, storage_key, entries, findings, projection) do
155 288 
    report = %{
156 289 
      repo: repo,
157 290 
      storage_key: storage_key,
291 
      node: node(),
158 292 
      entries: length(entries),
159 293 
      findings: findings,
160 294 
      head: head(entries),
161 
      chained_from: chained_from(entries)
295 
      chained_from: chained_from(entries),
296 
      status: status(findings, projection),
297 
      head_seq: projection.head_seq,
298 
      applied_seq: projection.applied_seq,
299 
      position: projection.position,
300 
      behind: projection.behind
162 301 
    }
163 302
164 303 
    if findings == [], do: {:ok, report}, else: {:error, report}
165 304 
  end
166 305
306 
  defp status([_finding | _rest], _projection), do: :diverged
307 
  defp status([], %{behind: 0}), do: :current
308 
  defp status([], _projection), do: :behind
309
167 310 
  defp head(entries) do
168 311 
    case List.last(entries) do
169 312 
      nil ->

@@ -258,36 +401,135 @@ defmodule OpenAgents.Forge.Verification do

258 401 
    [finding("entry_object_missing", %{"entry" => inspect(entry)})]
259 402 
  end
260 403
261 
  ## The served refs are the refs the WAL last recorded
404 
  ## Where on the log this node's projection sits
262 405
263 
  defp ref_findings(storage_key, index) do
264 
    recorded = index |> WAL.refs() |> Map.new()
406 
  # The projection's own marker says which entries it claims to have applied,
407 
  # and the served refs say what it is actually serving. The marker bounds the
408 
  # search; the refs decide it. A projection is *on* the log when it serves,
409 
  # exactly, the post-state some entry recorded — every ref that entry recorded
410 
  # and no other. The greatest such sequence at or above the marker is its
411 
  # position, and the distance from there to the head is how far behind it is.
412 
  #
413 
  # Everything below the marker is excluded because the projection has already
414 
  # claimed those entries: serving an older state than it claims to hold is a
415 
  # disagreement, not lag. Nothing above the head can be a candidate, so no
416 
  # marker can admit a state the log does not record.
417 
  defp locate(storage_key, entries) do
418 
    head_seq = head_seq(entries)
419 
    applied_seq = Repos.applied_seq(storage_key)
265 420 
    served = storage_key |> Repos.refs() |> Map.new()
421 
    position = position(entries, served, applied_seq, head_seq)
422
423 
    %{
424 
      head_seq: head_seq,
425 
      applied_seq: applied_seq,
426 
      served: served,
427 
      position: position,
428 
      behind: position && head_seq - position,
429 
      # The sequence whose state this projection is checked against: where it
430 
      # actually is, or — when it is nowhere on the log — where it claims to be.
431 
      checked_seq: position || min(applied_seq, head_seq)
432 
    }
433 
  end
266 434
267 
    diverged =
268 
      recorded
269 
      |> Map.keys()
270 
      |> Kernel.++(Map.keys(served))
271 
      |> Enum.uniq()
272 
      |> Enum.sort()
273 
      |> Enum.reject(fn name -> Map.get(recorded, name) == Map.get(served, name) end)
435 
  defp head_seq(entries) do
436 
    case List.last(entries) do
437 
      nil -> -1
438 
      entry -> entry["seq"]
439 
    end
440 
  end
441
442 
  defp position(entries, served, applied_seq, head_seq) do
443 
    on_log =
444 
      entries
445 
      |> Enum.filter(&(&1["seq"] >= applied_seq and &1["seq"] <= head_seq))
446 
      |> Enum.reverse()
447 
      |> Enum.find_value(fn entry -> if entry_refs(entry) == served, do: entry["seq"] end)
448
449 
    # Sequence -1 is the empty projection, which is every node before it
450 
    # replays anything. It is a candidate only when the marker claims nothing.
451 
    cond do
452 
      on_log != nil -> on_log
453 
      applied_seq <= -1 and served == %{} -> -1
454 
      true -> nil
455 
    end
456 
  end
457
458 
  defp entry_refs(entry) do
459 
    case entry && Map.get(entry, "refs") do
460 
      refs when is_map(refs) -> refs
461 
      _absent -> %{}
462 
    end
463 
  end
464
465 
  defp recorded_refs_at(_entries, seq) when seq < 0, do: %{}
466
467 
  defp recorded_refs_at(entries, seq) do
468 
    case Enum.find(entries, &(&1["seq"] == seq)) do
469 
      nil -> entries |> List.last() |> entry_refs()
470 
      entry -> entry_refs(entry)
471 
    end
472 
  end
473
474 
  ## The projection claims no more of the log than the log holds
475
476 
  # A marker past the end of the log is not lag in either direction: the
477 
  # projection says it applied an entry that is not there. A tail truncated out
478 
  # of the WAL leaves `0..n-1` contiguous and so passes every other check; this
479 
  # is where a node that had already applied past the truncation reports it.
480 
  defp marker_findings(%{applied_seq: applied_seq, head_seq: head_seq})
481 
       when applied_seq > head_seq do
482 
    [
483 
      finding("applied_seq_beyond_log", %{
484 
        "applied_seq" => applied_seq,
485 
        "head_seq" => head_seq
486 
      })
487 
    ]
488 
  end
489
490 
  defp marker_findings(_projection), do: []
491
492 
  ## The served refs are a state the WAL recorded
493
494 
  # A projection with a position is serving a state the log records, so there
495 
  # is nothing to report: it is current or behind, which `:status` says. One
496 
  # without a position is serving something the log never recorded at or after
497 
  # the sequence it claims, and every ref that differs from the claimed state
498 
  # is named.
499 
  defp ref_findings(_entries, %{position: position}) when position != nil, do: []
500
501 
  defp ref_findings(entries, projection) do
502 
    recorded = recorded_refs_at(entries, projection.checked_seq)
503 
    served = projection.served
274 504
275 
    Enum.map(diverged, fn name ->
505 
    recorded
506 
    |> Map.keys()
507 
    |> Kernel.++(Map.keys(served))
508 
    |> Enum.uniq()
509 
    |> Enum.sort()
510 
    |> Enum.reject(fn name -> Map.get(recorded, name) == Map.get(served, name) end)
511 
    |> Enum.map(fn name ->
276 512 
      finding("served_refs_diverged", %{
277 513 
        "ref" => name,
278 514 
        "recorded" => Map.get(recorded, name),
279 
        "served" => Map.get(served, name)
515 
        "served" => Map.get(served, name),
516 
        "at_seq" => projection.checked_seq
280 517 
      })
281 518 
    end)
282 519 
  end
283 520
284 
  ## Every object any accepted push named is present
521 
  ## Every object an applied push named is present
285 522
286 
  defp object_findings(storage_key, entries) do
523 
  # Bounded by where the projection is. An object introduced by an entry the
524 
  # node has not replayed is absent for the same reason the ref is, and both
525 
  # are the lag `:behind` reports. An object introduced at or below the
526 
  # projection's own position is not: it applied that entry.
527 
  defp object_findings(storage_key, entries, projection) do
287 528 
    path = Repos.bare_path(storage_key)
288 529
289 530 
    entries
290 
    |> Enum.flat_map(fn entry -> Map.to_list(entry["refs"] || %{}) end)
531 
    |> Enum.filter(&(&1["seq"] <= projection.checked_seq))
532 
    |> Enum.flat_map(fn entry -> Map.to_list(entry_refs(entry)) end)
291 533 
    |> Enum.uniq()
292 534 
    |> Enum.sort()
293 535 
    |> Enum.reject(fn {_name, sha} ->

@@ -312,12 +554,16 @@ defmodule OpenAgents.Forge.Verification do

312 554 
  # A shallow graft the repository legitimately carries stops the walk at its
313 555 
  # boundary, so a grafted repository is clean here: it is servable, and
314 556 
  # servable is the claim.
315 
  defp reachability_findings(storage_key, index) do
557 
  #
558 
  # The tips are the ones recorded at the projection's position, because a
559 
  # clone answered by this node walks the state this node holds, not the state
560 
  # the head describes.
561 
  defp reachability_findings(storage_key, entries, projection) do
316 562 
    path = Repos.bare_path(storage_key)
317 563
318 564 
    tips =
319 
      index
320 
      |> WAL.refs()
565 
      entries
566 
      |> recorded_refs_at(projection.checked_seq)
321 567 
      |> exportable_refs()
322 568 
      |> Map.values()
323 569 
      |> Enum.uniq()

@@ -433,6 +679,173 @@ defmodule OpenAgents.Forge.Verification do

433 679
434 680 
  defp normalize_anchor(_absent_or_malformed), do: nil
435 681
682 
  @doc """
683 
  Verify one repository across every node that serves it.
684
685 
  Each member answers for its own projection, so the fleet's answer is the
686 
  combination of theirs. The verdict is deliberately insensitive to lag:
687
688 
  * `:verified` — every member answered and every one is at the head.
689 
  * `:converging` — nothing contradicts the log, and at least one member is
690 
    behind it or did not answer. This state clears itself as nodes replay.
691 
  * `:diverged` — a member's projection contradicts the log, or two members
692 
    disagree about the log itself. Nothing here clears itself.
693 
  * `:unavailable` — nothing was checked: no member answered, or every member
694 
    could not resolve the repository or read its log.
695
696 
  Returns `{:ok, cluster_report}` for `:verified` and `:converging`, and
697 
  `{:error, cluster_report}` for `:diverged` and `:unavailable`. Findings are
698 
  carried with the node that produced them, so a reader can tell one node's
699 
  answer from the fleet's.
700
701 
  Options:
702
703 
  * `:anchor` — passed through to `verify/2` on every member. Because the
704 
    anchor comes from outside the log, this is also how the members are checked
705 
    against one repository history rather than only against each other.
706 
  * `:members` — a zero-arity function returning the members to ask. Defaults
707 
    to `OpenAgents.Cluster.members/0`.
708 
  * `:rpc` — a five-arity function with `:erpc.call/5`'s shape.
709 
  * `:timeout_ms` — per-member deadline.
710
711 
  This reaches no database either: membership comes from `OpenAgents.Cluster`,
712 
  which reads `Node`, and each member runs the same WAL-and-repository check.
713
714 
  `:log_agreement` compares the members that reported a chain link at the same
715 
  sequence. Two nodes reading one shared log cannot disagree there, so
716 
  `:disagreed` means one of them is not reading the log the other is. Members
717 
  at different sequences have nothing comparable, and the value is `:unknown`
718 
  when no two members reported the same one.
719 
  """
720 
  @spec verify_cluster(RepoRef.ref(), keyword()) ::
721 
          {:ok, cluster_report()} | {:error, cluster_report()}
722 
  def verify_cluster(repo_ref, opts \\ []) when is_binary(repo_ref) and is_list(opts) do
723 
    members = Keyword.get(opts, :members, &Cluster.members/0).() |> Enum.uniq()
724 
    rpc = Keyword.get(opts, :rpc, &:erpc.call/5)
725 
    timeout_ms = Keyword.get(opts, :timeout_ms, @cluster_timeout_ms)
726 
    verify_opts = Keyword.take(opts, [:anchor])
727
728 
    results =
729 
      members
730 
      |> Task.async_stream(
731 
        fn member -> ask(member, repo_ref, verify_opts, rpc, timeout_ms) end,
732 
        ordered: true,
733 
        timeout: timeout_ms + 1_000,
734 
        on_timeout: :kill_task,
735 
        max_concurrency: max(1, length(members))
736 
      )
737 
      |> Enum.zip(members)
738 
      |> Enum.map(fn
739 
        {{:ok, result}, member} -> node_result(member, result)
740 
        {{:exit, reason}, member} -> unreachable_result(member, reason)
741 
      end)
742
743 
    cluster_report(repo_ref, results)
744 
  end
745
746 
  defp ask(member, repo_ref, verify_opts, _rpc, _timeout_ms) when member == node(),
747 
    do: verify(repo_ref, verify_opts)
748
749 
  defp ask(member, repo_ref, verify_opts, rpc, timeout_ms) do
750 
    rpc.(member, __MODULE__, :verify, [repo_ref, verify_opts], timeout_ms)
751 
  catch
752 
    kind, reason -> {:unreachable, {kind, reason}}
753 
  end
754
755 
  defp node_result(member, {ok_or_error, report}) when ok_or_error in [:ok, :error] do
756 
    %{
757 
      node: member,
758 
      status: report.status,
759 
      head_seq: report.head_seq,
760 
      applied_seq: report.applied_seq,
761 
      position: report.position,
762 
      behind: report.behind,
763 
      entries: report.entries,
764 
      head: report.head,
765 
      findings: report.findings,
766 
      reason: nil
767 
    }
768 
  end
769
770 
  defp node_result(member, {:unreachable, reason}), do: unreachable_result(member, reason)
771 
  defp node_result(member, other), do: unreachable_result(member, {:invalid_result, other})
772
773 
  defp unreachable_result(member, reason) do
774 
    %{
775 
      node: member,
776 
      status: :unreachable,
777 
      head_seq: nil,
778 
      applied_seq: nil,
779 
      position: nil,
780 
      behind: nil,
781 
      entries: nil,
782 
      head: nil,
783 
      findings: [],
784 
      reason: reason
785 
    }
786 
  end
787
788 
  defp cluster_report(repo_ref, results) do
789 
    agreement = log_agreement(results)
790
791 
    findings =
792 
      Enum.flat_map(results, fn result ->
793 
        Enum.map(result.findings, &Map.put(&1, :node, result.node))
794 
      end)
795
796 
    report = %{
797 
      repo: repo_ref,
798 
      status: cluster_status(results, agreement),
799 
      head_seq: head_seq_across(results),
800 
      log_agreement: agreement,
801 
      nodes: results,
802 
      findings: findings
803 
    }
804
805 
    if report.status in [:verified, :converging], do: {:ok, report}, else: {:error, report}
806 
  end
807
808 
  defp head_seq_across(results) do
809 
    case results |> Enum.map(& &1.head_seq) |> Enum.reject(&is_nil/1) do
810 
      [] -> nil
811 
      seqs -> Enum.max(seqs)
812 
    end
813 
  end
814
815 
  defp cluster_status(results, agreement) do
816 
    checked = Enum.filter(results, &(&1.status in [:current, :behind, :diverged]))
817
818 
    cond do
819 
      Enum.any?(results, &(&1.status == :diverged)) -> :diverged
820 
      agreement == :disagreed -> :diverged
821 
      checked == [] -> :unavailable
822 
      # Some member cannot see a repository the others verified, which is a
823 
      # disagreement about the fleet rather than a lag window.
824 
      Enum.any?(results, &(&1.status == :unresolved)) -> :diverged
825 
      Enum.all?(results, &(&1.status == :current)) -> :verified
826 
      true -> :converging
827 
    end
828 
  end
829
830 
  # Two nodes that reported a link for the same sequence must have reported the
831 
  # same link: the WAL is one shared log. Nodes at different sequences are
832 
  # compared on nothing, which is why lag cannot reach this value.
833 
  defp log_agreement(results) do
834 
    by_seq =
835 
      results
836 
      |> Enum.map(& &1.head)
837 
      |> Enum.reject(&is_nil/1)
838 
      |> Enum.group_by(& &1.seq, & &1.link)
839
840 
    comparable = Enum.filter(by_seq, fn {_seq, links} -> length(links) > 1 end)
841
842 
    cond do
843 
      comparable == [] -> :unknown
844 
      Enum.all?(comparable, fn {_seq, links} -> Enum.uniq(links) |> length() == 1 end) -> :agreed
845 
      true -> :disagreed
846 
    end
847 
  end
848
436 849 
  @doc """
437 850 
  The refs a clone receives: every recorded ref except the hidden internal
438 851 
  bookkeeping namespace.
test/openagents/forge/independence_test.exs modified +13 -4

@@ -162,13 +162,22 @@ defmodule OpenAgents.Forge.IndependenceTest do

162 162 
      assert %{"seq" => 0} = detail(findings, "entry_object_missing")
163 163 
    end
164 164
165 
    test "a lost cache is reported as diverged refs and missing objects", context do
165 
    test "a lost cache is reported as an empty projection, not as a disagreement", context do
166 166 
      seed_history!(context)
167 
      {:ok, %{status: :current, head_seq: head_seq}} = Verification.verify(context.repo)
167 168 
      File.rm_rf!(Repos.bare_path(context.repo))
168 169
169 
      assert {:error, %{findings: findings}} = Verification.verify(context.repo)
170 
      assert detail(findings, "served_refs_diverged") != nil
171 
      assert detail(findings, "object_missing") != nil
170 
      # A cache that was deleted and a node that has not replayed yet are the
171 
      # same observation from the WAL and the repository alone, and the second
172 
      # is every node's ordinary state (issue #251). Both are reported as an
173 
      # empty projection at sequence -1 rather than as tampering, which is what
174 
      # `REPOSITORY-003` says a disposable projection is. It is still not
175 
      # reported as current: the report says how far from the log it is.
176 
      assert {:ok, report} = Verification.verify(context.repo)
177 
      assert report.findings == []
178 
      assert report.status == :behind
179 
      assert report.position == -1
180 
      assert report.behind == head_seq + 1
172 181 
    end
173 182
174 183 
    test "verification reaches no database", _context do
test/openagents/forge/verification_test.exs added +389

@@ -0,0 +1,389 @@

1 
defmodule OpenAgents.Forge.VerificationTest do
2 
  @moduledoc """
3 
  EXIT-002, the part that only appears on a fleet: three nodes share one WAL
4 
  and keep three separate projections of it, so at any moment they are at
5 
  three different sequences.
6
7 
  Every node here is a real bare repository built by real replay from a real
8 
  WAL whose entries are genuine `receive-pack` requests. The nodes differ only
9 
  in `:forge_data_dir`, which is exactly how they differ in production: the WAL
10 
  is shared object storage and `/var/lib/openagents/forge/repos` is local disk.
11 
  """
12
13 
  use OpenAgents.DataCase, async: false
14
15 
  alias OpenAgents.Forge.{Repos, Sync, Verification, WAL}
16
17 
  defmodule TestPipeline do
18 
    @moduledoc false
19 
    use Plug.Builder
20
21 
    plug OpenAgentsWeb.Plugs.ForgeGitAuth
22 
    plug OpenAgents.Forge.GitHTTP
23 
  end
24
25 
  setup do
26 
    Ecto.Adapters.SQL.Sandbox.mode(OpenAgents.Repo, {:shared, self()})
27
28 
    base =
29 
      Path.join(System.tmp_dir!(), "forge-verification-#{System.unique_integer([:positive])}")
30
31 
    File.mkdir_p!(base)
32 
    previous_data = Application.get_env(:openagents, :forge_data_dir)
33 
    previous_wal = Application.get_env(:openagents, :forge_wal_dir)
34
35 
    # One WAL, three data directories: the fleet's actual shape.
36 
    Application.put_env(:openagents, :forge_wal_dir, Path.join(base, "wal"))
37 
    Application.put_env(:openagents, :forge_data_dir, node_dir(base, :one))
38 
    OpenAgents.Forge.CacheReadiness.reset()
39
40 
    user = OpenAgents.AccountsFixtures.repository_user_fixture("fleet-owner")
41
42 
    {:ok, repository, :created} =
43 
      OpenAgents.Repositories.create_user_repository(user, %{name: "demo"}, "fleet-demo")
44
45 
    repository =
46 
      repository
47 
      |> Ecto.Changeset.change(lifecycle_state: "ready", ready_at: DateTime.utc_now())
48 
      |> OpenAgents.Repo.update!()
49
50 
    {:ok, _api_token, plaintext} =
51 
      OpenAgents.ApiTokens.create(user, %{
52 
        name: "forge verification test",
53 
        scopes: ["forge:write"],
54 
        lifetime_days: 1
55 
      })
56
57 
    port = free_port()
58 
    start_supervised!({Bandit, plug: TestPipeline, port: port, ip: {127, 0, 0, 1}})
59
60 
    on_exit(fn ->
61 
      Application.put_env(:openagents, :forge_data_dir, previous_data)
62 
      Application.put_env(:openagents, :forge_wal_dir, previous_wal)
63 
      OpenAgents.Forge.CacheReadiness.reset()
64 
      File.rm_rf(base)
65 
    end)
66
67 
    %{
68 
      base: base,
69 
      repo: repository.storage_key,
70 
      repository: repository,
71 
      # `verify_cluster/2` asks its members concurrently, and in production each
72 
      # member is a different machine with its own `:forge_data_dir`. Here they
73 
      # are one VM sharing one application environment, so the stand-in for a
74 
      # node runs through this agent, which serializes what distribution would
75 
      # have separated.
76 
      nodes: start_supervised!({Agent, fn -> :ok end}),
77 
      url: "http://x:#{plaintext}@127.0.0.1:#{port}/fleet-owner/demo.git"
78 
    }
79 
  end
80
81 
  describe "a node that has not replayed yet" do
82 
    test "is reported as behind rather than as a disagreement", context do
83 
      seed_history!(context)
84 
      replay!(context, :two)
85
86 
      # One more accepted push. Node one applied it; node two has not read the
87 
      # log since, which is every node's ordinary state for the minute after a
88 
      # push lands somewhere else.
89 
      commit_and_push!(work_dir(context), "later.txt", "later\n", "later")
90
91 
      assert {:ok, current} = at_node(context, :one, fn -> Verification.verify(context.repo) end)
92 
      assert current.status == :current
93 
      assert current.behind == 0
94
95 
      assert {:ok, behind} = at_node(context, :two, fn -> Verification.verify(context.repo) end)
96 
      assert behind.findings == []
97 
      assert behind.status == :behind
98 
      assert behind.behind == 1
99 
      assert behind.position == current.position - 1
100 
      assert behind.applied_seq == behind.position
101 
      assert behind.head_seq == current.head_seq
102 
    end
103
104 
    test "names the node and the sequence its answer was computed at", context do
105 
      seed_history!(context)
106 
      replay!(context, :two)
107
108 
      assert {:ok, report} = at_node(context, :two, fn -> Verification.verify(context.repo) end)
109 
      assert report.node == node()
110 
      assert report.applied_seq == report.head_seq
111 
      assert report.position == report.head_seq
112 
    end
113
114 
    test "an empty projection is the same answer at sequence -1", context do
115 
      seed_history!(context)
116
117 
      assert {:ok, report} = at_node(context, :three, fn -> Verification.verify(context.repo) end)
118 
      assert report.findings == []
119 
      assert report.status == :behind
120 
      assert report.position == -1
121 
      assert report.applied_seq == -1
122 
      assert report.behind == report.head_seq + 1
123 
    end
124
125 
    test "catches up to clean once it replays", context do
126 
      seed_history!(context)
127
128 
      assert {:ok, %{status: :behind}} =
129 
               at_node(context, :two, fn -> Verification.verify(context.repo) end)
130
131 
      replay!(context, :two)
132
133 
      assert {:ok, %{status: :current, behind: 0, findings: []}} =
134 
               at_node(context, :two, fn -> Verification.verify(context.repo) end)
135 
    end
136 
  end
137
138 
  describe "a projection that contradicts the log" do
139 
    test "a ref moved without a push is reported on a current node", context do
140 
      seed_history!(context)
141 
      path = at_node(context, :one, fn -> Repos.bare_path(context.repo) end)
142 
      {parent, 0} = Repos.git(path, ["rev-parse", "refs/heads/main^"])
143 
      {_output, 0} = Repos.git(path, ["update-ref", "refs/heads/main", String.trim(parent)])
144
145 
      assert {:error, report} =
146 
               at_node(context, :one, fn -> Verification.verify(context.repo) end)
147
148 
      assert report.status == :diverged
149 
      assert %{"ref" => "refs/heads/main"} = detail(report.findings, "served_refs_diverged")
150 
    end
151
152 
    test "a ref moved without a push is reported on a node that is also behind", context do
153 
      seed_history!(context)
154 
      replay!(context, :two)
155 
      commit_and_push!(work_dir(context), "later.txt", "later\n", "later")
156
157 
      # Node two is a legitimate entry behind *and* someone moved a ref on its
158 
      # disk. Lag must not launder the second fact.
159 
      path = at_node(context, :two, fn -> Repos.bare_path(context.repo) end)
160 
      {parent, 0} = Repos.git(path, ["rev-parse", "refs/heads/main^"])
161 
      {_output, 0} = Repos.git(path, ["update-ref", "refs/heads/main", String.trim(parent)])
162
163 
      assert {:error, report} =
164 
               at_node(context, :two, fn -> Verification.verify(context.repo) end)
165
166 
      assert report.status == :diverged
167 
      assert %{"ref" => "refs/heads/main"} = detail(report.findings, "served_refs_diverged")
168 
    end
169
170 
    test "a ref the log never recorded is reported however far behind the node is", context do
171 
      seed_history!(context)
172 
      replay!(context, :two)
173 
      commit_and_push!(work_dir(context), "later.txt", "later\n", "later")
174
175 
      path = at_node(context, :two, fn -> Repos.bare_path(context.repo) end)
176 
      {head, 0} = Repos.git(path, ["rev-parse", "refs/heads/main"])
177 
      {_output, 0} = Repos.git(path, ["update-ref", "refs/heads/smuggled", String.trim(head)])
178
179 
      assert {:error, report} =
180 
               at_node(context, :two, fn -> Verification.verify(context.repo) end)
181
182 
      assert %{"ref" => "refs/heads/smuggled", "recorded" => nil} =
183 
               detail(report.findings, "served_refs_diverged")
184 
    end
185
186 
    test "an object an applied entry introduced cannot be missing", context do
187 
      seed_history!(context)
188 
      path = at_node(context, :one, fn -> Repos.bare_path(context.repo) end)
189 
      {head, 0} = Repos.git(path, ["rev-parse", "refs/heads/feature"])
190 
      head = String.trim(head)
191
192 
      File.rm_rf!(Path.join([path, "objects", String.slice(head, 0, 2)]))
193 
      {_output, 0} = Repos.git(path, ["update-ref", "-d", "refs/heads/feature"])
194
195 
      assert {:error, report} =
196 
               at_node(context, :one, fn -> Verification.verify(context.repo) end)
197
198 
      assert %{"object" => ^head} = detail(report.findings, "object_missing")
199 
    end
200
201 
    test "a marker naming a sequence the log does not have is reported", context do
202 
      seed_history!(context)
203 
      path = at_node(context, :one, fn -> Repos.bare_path(context.repo) end)
204 
      Repos.record_applied_seq_at!(path, 99)
205
206 
      assert {:error, report} =
207 
               at_node(context, :one, fn -> Verification.verify(context.repo) end)
208
209 
      assert %{"applied_seq" => 99} = detail(report.findings, "applied_seq_beyond_log")
210 
    end
211
212 
    test "a marker rolled forward does not silence a stale projection", context do
213 
      seed_history!(context)
214 
      replay!(context, :two)
215 
      commit_and_push!(work_dir(context), "later.txt", "later\n", "later")
216
217 
      # The marker is the projection's own claim about itself. Claiming to have
218 
      # applied the head does not make the head's refs appear.
219 
      path = at_node(context, :two, fn -> Repos.bare_path(context.repo) end)
220 
      {:ok, _generation, index} = at_node(context, :two, fn -> WAL.read_index(context.repo) end)
221 
      Repos.record_applied_seq_at!(path, WAL.next_seq(index) - 1)
222
223 
      assert {:error, report} =
224 
               at_node(context, :two, fn -> Verification.verify(context.repo) end)
225
226 
      assert detail(report.findings, "served_refs_diverged") != nil
227 
    end
228 
  end
229
230 
  describe "the whole fleet at once" do
231 
    test "three nodes at three sequences converge rather than disagree", context do
232 
      seed_history!(context)
233 
      replay!(context, :two)
234 
      replay!(context, :three)
235 
      commit_and_push!(work_dir(context), "one-later.txt", "one\n", "one later")
236 
      replay!(context, :two)
237 
      commit_and_push!(work_dir(context), "two-later.txt", "two\n", "two later")
238
239 
      assert {:ok, cluster} = verify_cluster(context)
240 
      assert cluster.status == :converging
241 
      assert cluster.findings == []
242 
      assert Enum.map(cluster.nodes, & &1.behind) == [0, 1, 2]
243 
      assert Enum.map(cluster.nodes, & &1.status) == [:current, :behind, :behind]
244 
      assert Enum.uniq(Enum.map(cluster.nodes, & &1.head_seq)) == [cluster.head_seq]
245 
      assert cluster.log_agreement == :agreed
246 
    end
247
248 
    test "a fleet that has finished replaying is verified", context do
249 
      seed_history!(context)
250 
      replay!(context, :two)
251 
      replay!(context, :three)
252
253 
      assert {:ok, cluster} = verify_cluster(context)
254 
      assert cluster.status == :verified
255 
      assert Enum.all?(cluster.nodes, &(&1.status == :current))
256 
    end
257
258 
    test "one node's tampering is the fleet's answer, and it names the node", context do
259 
      seed_history!(context)
260 
      replay!(context, :two)
261 
      replay!(context, :three)
262
263 
      path = at_node(context, :three, fn -> Repos.bare_path(context.repo) end)
264 
      {head, 0} = Repos.git(path, ["rev-parse", "refs/heads/main"])
265 
      {_output, 0} = Repos.git(path, ["update-ref", "refs/heads/smuggled", String.trim(head)])
266
267 
      assert {:error, cluster} = verify_cluster(context)
268 
      assert cluster.status == :diverged
269 
      assert [%{node: :three, code: "served_refs_diverged"}] = cluster.findings
270 
      assert Enum.map(cluster.nodes, & &1.status) == [:current, :current, :diverged]
271 
    end
272
273 
    test "a node that does not answer leaves the fleet converging, not verified", context do
274 
      seed_history!(context)
275 
      replay!(context, :two)
276 
      replay!(context, :three)
277
278 
      rpc = fn
279 
        :three, _module, _function, _args, _timeout ->
280 
          exit(:noconnection)
281
282 
        member, module, function, args, _timeout ->
283 
          apply_at(context, member, module, function, args)
284 
      end
285
286 
      assert {:ok, cluster} = verify_cluster(context, rpc: rpc)
287 
      assert cluster.status == :converging
288 
      assert cluster.findings == []
289 
      assert %{node: :three, status: :unreachable} = List.last(cluster.nodes)
290 
    end
291
292 
    test "no node answering is not a clean fleet", context do
293 
      seed_history!(context)
294
295 
      rpc = fn _member, _module, _function, _args, _timeout -> exit(:noconnection) end
296
297 
      assert {:error, cluster} = verify_cluster(context, rpc: rpc)
298 
      assert cluster.status == :unavailable
299 
    end
300 
  end
301
302 
  ## ── helpers ────────────────────────────────────────────────────────────
303
304 
  # Three nodes, one WAL. `:erpc` is replaced by a call that runs the same
305 
  # verification against the named node's own data directory, which is the one
306 
  # thing that actually differs between fleet members.
307 
  defp verify_cluster(context, options \\ []) do
308 
    rpc =
309 
      Keyword.get(options, :rpc, fn member, module, function, args, _timeout ->
310 
        apply_at(context, member, module, function, args)
311 
      end)
312
313 
    Verification.verify_cluster(context.repo, members: fn -> [:one, :two, :three] end, rpc: rpc)
314 
  end
315
316 
  defp apply_at(context, member, module, function, args) do
317 
    Agent.get(
318 
      context.nodes,
319 
      fn _state -> at_node(context, member, fn -> apply(module, function, args) end) end,
320 
      30_000
321 
    )
322 
  end
323
324 
  defp at_node(context, member, function) do
325 
    previous = Application.get_env(:openagents, :forge_data_dir)
326 
    Application.put_env(:openagents, :forge_data_dir, node_dir(context.base, member))
327
328 
    try do
329 
      function.()
330 
    after
331 
      Application.put_env(:openagents, :forge_data_dir, previous)
332 
    end
333 
  end
334
335 
  defp node_dir(base, member), do: Path.join(base, "data-#{member}")
336
337 
  defp replay!(context, member) do
338 
    assert :ok = at_node(context, member, fn -> Sync.ensure_fresh(context.repo) end)
339 
  end
340
341 
  defp detail(findings, code) do
342 
    Enum.find_value(findings, fn
343 
      %{code: ^code, detail: detail} -> detail
344 
      _other -> nil
345 
    end)
346 
  end
347
348 
  defp work_dir(context), do: Path.join(context.base, "work")
349
350 
  defp seed_history!(context) do
351 
    work = work_dir(context)
352
353 
    unless File.exists?(work) do
354 
      sh!(context.base, "git", ["clone", context.url, work])
355 
      sh!(work, "git", ["config", "user.email", "test@example.com"])
356 
      sh!(work, "git", ["config", "user.name", "Forge Test"])
357 
      commit_and_push!(work, "one.txt", "one\n", "one")
358 
      commit_and_push!(work, "two.txt", "two\n", "two")
359 
      sh!(work, "git", ["checkout", "-b", "feature"])
360 
      commit_and_push!(work, "feature.txt", "feature\n", "feature", "feature")
361 
      sh!(work, "git", ["checkout", "main"])
362 
    end
363
364 
    :ok
365 
  end
366
367 
  defp commit_and_push!(work, filename, contents, message, branch \\ "main") do
368 
    File.write!(Path.join(work, filename), contents)
369 
    sh!(work, "git", ["add", "."])
370 
    sh!(work, "git", ["commit", "-m", message])
371 
    sh!(work, "git", ["push", "origin", "HEAD:#{branch}"])
372 
  end
373
374 
  defp free_port do
375 
    {:ok, socket} = :gen_tcp.listen(0, [])
376 
    {:ok, port} = :inet.port(socket)
377 
    :gen_tcp.close(socket)
378 
    port
379 
  end
380
381 
  defp sh!(dir, "git", args), do: sh_raw!(dir, "git", ["-c", "credential.helper="] ++ args)
382 
  defp sh!(dir, command, args), do: sh_raw!(dir, command, args)
383
384 
  defp sh_raw!(dir, command, args) do
385 
    {output, status} = System.cmd(command, args, cd: dir, stderr_to_stdout: true)
386 
    if status != 0, do: flunk("#{command} #{Enum.join(args, " ")} failed:\n#{output}")
387 
    output
388 
  end
389 
end

This page updates live while a promote is in flight · changelog