coverage — ~bigbes/sr-ht-spec — 3cb1c03d — service/reconcile.go

Coverage
82.3% 116/141 statements
Δ
—
Blob
ee2a26d
Uncovered L99-L101L237L256-L260L306-L307L312-L313L317-L318L326-L327L337-L338L346-L350L375-L376L379-L380L403-L404L419-L421L448-L449L462-L463L506-L507L522-L523
1 package service
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "time"
8
9 "github.com/go-git/go-git/v5/plumbing"
10
11 "sourcecraft.dev/bigbes/sr-ht-spec/core"
12 "sourcecraft.dev/bigbes/sr-ht-spec/db"
13 "sourcecraft.dev/bigbes/sr-ht-spec/gitx"
14 )
15
16 // Three systems are touched by a merge — git refs, the bleve index and
17 // Postgres — and none of it is transactional. The rule that makes it tractable:
18 //
19 // Git refs are the source of truth for whether a proposal has merged. The
20 // Postgres row is the source of truth for that a proposal exists and what it
21 // is. The index and the render cache are pure caches.
22 //
23 // The reconciler is the backstop that repairs divergence, and it implements
24 // exactly four repairs — no more, because every additional guess about what a
25 // half-finished write meant is a way to invent state nobody wrote.
26
27 const (
28 // DefaultReconcileInterval is how often the reconciler runs after startup.
29 DefaultReconcileInterval = 15 * time.Minute
30
31 // DefaultReconcileGrace is how long a proposal row with no branch is left
32 // alone before it is deleted.
33 //
34 // The design's repair table has no grace period, and without one the
35 // reconciler is actively destructive during ordinary operation: a proposal
36 // is opened row-first, so every live propose passes through the exact state
37 // ("open row, no branch") that the table says to delete. The window is
38 // milliseconds wide and the reconciler runs on a timer, so it would be rare
39 // — which makes it worse, not better, since it would destroy an agent's
40 // work at random and never in a test.
41 //
42 // A row younger than this is therefore assumed to be in flight rather than
43 // abandoned. It costs one extra reconcile cycle before a genuinely crashed
44 // proposal is cleaned up, which nothing is waiting on.
45 DefaultReconcileGrace = 5 * time.Minute
46 )
47
48 // RepairKind names one of the four repairs.
49 type RepairKind string
50
51 const (
52 // RepairDeleteRow removes an `open` proposal row whose branch does not
53 // exist: the daemon died between the row insert and the branch write. The
54 // row holds no content, and the agent still holds the document it wanted to
55 // write, so it re-proposes.
56 RepairDeleteRow RepairKind = "delete-proposal-row"
57
58 // RepairDeleteRef removes a proposals/* ref with no row. It is unreferenced
59 // — the id is a Postgres serial, and title, rationale, base_rev, agent and
60 // agent_session live nowhere in a ref — so its content is unrecoverable
61 // anyway and recreating the row would mean inventing every field.
62 RepairDeleteRef RepairKind = "delete-proposal-ref"
63
64 // RepairMarkMerged transitions a row still `open` whose branch has merged
65 // into the approved head. The ref is truth for merged-ness.
66 RepairMarkMerged RepairKind = "mark-proposal-merged"
67
68 // RepairReindex flags a space whose index stamp differs from its approved
69 // head. Phase 2 owns the rebuild; the reconciler only reports the list.
70 RepairReindex RepairKind = "reindex-space"
71 )
72
73 // Repair is one repair the reconciler decided on.
74 type Repair struct {
75 Kind RepairKind
76 Space core.SpaceRef
77 SpaceID int
78
79 // ProposalID is the proposal row's id, and zero for RepairReindex and for
80 // an orphan ref whose name carries no parseable id.
81 ProposalID int
82
83 // Branch is the proposal branch this repair is about, empty for
84 // RepairReindex.
85 Branch string
86
87 // Rev is the revision the repair records: the merge revision for
88 // RepairMarkMerged, the approved head for RepairReindex.
89 Rev string
90
91 // Approval is the approval kind RepairMarkMerged records. See
92 // PlanRepairs for why it is always core.ApprovalPolicy.
93 Approval core.Approval
94
95 // Reason is a human-readable sentence for the log.
96 Reason string
97 }
98
99 0 func (r Repair) String() string {
100 0 return fmt.Sprintf("%s %s: %s", r.Kind, r.Space, r.Reason)
101 0 }
102
103 // ProposalFact is what the reconciler observed about one proposal — its row,
104 // its branch, or both. Facts are gathered by I/O and consumed by PlanRepairs,
105 // which is pure so that the decision table can be exhaustively tested without a
106 // repository or a database.
107 type ProposalFact struct {
108 // ID is the proposal row id, and the id parsed out of the branch name when
109 // there is no row. Zero when the branch name carries no parseable id.
110 ID int
111
112 // Branch is the proposal branch name, "proposals/42".
113 Branch string
114
115 // HasRow and HasBranch record which halves exist. Both false is not a fact.
116 HasRow bool
117 HasBranch bool
118
119 // State is the row's state, meaningless when HasRow is false.
120 State core.ProposalState
121
122 // Created is when the row was inserted, for the grace window.
123 Created time.Time
124
125 // MergedIntoApproved reports whether the branch tip is reachable from the
126 // approved head. Computed by I/O (it needs the object database) and passed
127 // in, exactly as gitx.RefUpdate.FastForward is.
128 MergedIntoApproved bool
129
130 // BranchHead is the branch tip.
131 BranchHead string
132
133 // BaseRev is the proposal's recorded base, resolved to an object name, and
134 // empty when it could not be resolved. See PlanRepairs for why a branch
135 // still sitting on its base is not a merge.
136 BaseRev string
137 }
138
139 // SpaceFacts is everything the reconciler observed about one space.
140 type SpaceFacts struct {
141 Space core.SpaceRef
142 SpaceID int
143
144 // ApprovedHead is the current tip of the approved branch.
145 ApprovedHead string
146
147 // IndexRev is the revision the global index currently reflects for this
148 // space, empty when the space has never been indexed. Empty is stale by
149 // construction, which is why a missing stamp is not treated as up to date.
150 IndexRev string
151
152 Proposals []ProposalFact
153
154 // Now and Grace parameterize the grace window, so it is an input to the
155 // decision rather than a clock read inside it.
156 Now time.Time
157 Grace time.Duration
158 }
159
160 // PlanRepairs is the repair table, as a pure function.
161 //
162 // It implements exactly the four rows of the design's "Consistency and
163 // recovery" table and nothing else. States it does not name — a merged row
164 // whose branch is no longer an ancestor of the approved head, a rejected row
165 // whose branch still exists — are deliberately left alone: neither is a
166 // half-finished write, and repairing them would mean deciding something the
167 // design did not.
168 //
169 // "Merged" needs one qualification the design's table does not state, and
170 // without it the reconciler corrupts state during ordinary operation. A
171 // proposal branch is cut *at* the approved head, so between the cut and the
172 // agent's first commit its tip is trivially an ancestor of that head — and the
173 // literal rule "branch merged into the approved head, row still open" fires on
174 // a proposal that has not merged and has no content at all. The same holds
175 // forever after for a proposal whose agent never committed. A branch still
176 // sitting on its recorded base is therefore never treated as merged, and a base
177 // that could not be resolved is treated the same way: repairing on facts we
178 // could not establish is worse than leaving the row open for a human to see.
179 //
180 // RepairMarkMerged always records core.ApprovalPolicy. The ref proves the
181 // merge happened and nothing proves how it was authorized — the approval kind
182 // existed only in the memory of the process that died. Of the two available
183 // lies, "policy" is the safe one: recording "human" would launder unreviewed
184 // content as blessed, which is the exact failure the bimodal decision exists to
185 // prevent, while recording "policy" understates the review and puts the
186 // proposal in the policy-merged digest, where a human sees it again. Erring
187 // toward visibility is the whole point of the digest.
188 20 func PlanRepairs(f SpaceFacts) []Repair {
189 20 var repairs []Repair
190 20 base := Repair{Space: f.Space, SpaceID: f.SpaceID}
191 20
192 22 for _, p := range f.Proposals {
193 22 r := base
194 22 r.ProposalID = p.ID
195 22 r.Branch = p.Branch
196 22
197 22 switch {
198 case p.HasRow && p.HasBranch && p.State == core.StateOpen && p.MergedIntoApproved &&
199 4 p.BaseRev != "" && p.BranchHead != p.BaseRev:
200 4 r.Kind = RepairMarkMerged
201 4 r.Rev = f.ApprovedHead
202 4 r.Approval = core.ApprovalPolicy
203 4 r.Reason = fmt.Sprintf("branch %s has merged into the approved head %s but the row is still open",
204 4 p.Branch, short(f.ApprovedHead))
205 4 repairs = append(repairs, r)
206
207 5 case p.HasRow && !p.HasBranch && p.State == core.StateOpen:
208 5 if f.Now.Sub(p.Created) < f.Grace {
209 2 continue // in flight: the row is written before the branch
210 }
211 3 r.Kind = RepairDeleteRow
212 3 r.Reason = fmt.Sprintf("row is open but branch %s does not exist; the agent re-proposes", p.Branch)
213 3 repairs = append(repairs, r)
214
215 4 case !p.HasRow && p.HasBranch:
216 4 r.Kind = RepairDeleteRef
217 4 r.Reason = fmt.Sprintf("branch %s has no row; its content is unrecoverable", p.Branch)
218 4 repairs = append(repairs, r)
219 }
220 }
221
222 20 if f.ApprovedHead != "" && f.IndexRev != f.ApprovedHead {
223 6 r := base
224 6 r.Kind = RepairReindex
225 6 r.Rev = f.ApprovedHead
226 6 r.Reason = fmt.Sprintf("index stamp %s differs from the approved head %s",
227 6 stampOrNever(f.IndexRev), short(f.ApprovedHead))
228 6 repairs = append(repairs, r)
229 6 }
230 20 return repairs
231 }
232
233 16 func short(rev string) string {
234 16 if len(rev) > 8 {
235 16 return rev[:8]
236 16 }
237 0 return rev
238 }
239
240 6 func stampOrNever(rev string) string {
241 6 if rev == "" {
242 4 return "(never indexed)"
243 4 }
244 2 return short(rev)
245 }
246
247 // ReconcileFailure is one thing the reconciler could not do. Failures never
248 // abort the run: a space with an unreadable repository must not stop the other
249 // spaces from being repaired.
250 type ReconcileFailure struct {
251 Space core.SpaceRef
252 Repair *Repair // nil when the whole space could not be examined
253 Err error
254 }
255
256 0 func (f ReconcileFailure) Error() string {
257 0 if f.Repair != nil {
258 0 return fmt.Sprintf("%s: %v", f.Repair, f.Err)
259 0 }
260 0 return fmt.Sprintf("%s: %v", f.Space, f.Err)
261 }
262
263 // ReconcileReport is what one reconciler pass did.
264 type ReconcileReport struct {
265 // Spaces is how many spaces were examined.
266 Spaces int
267
268 // Repaired lists the repairs that were applied.
269 Repaired []Repair
270
271 // Reindex lists spaces whose index is stale. They are reported, not
272 // repaired: bleve is single-writer and Phase 2 owns the index. A caller
273 // that has an indexer drives it from this list.
274 Reindex []Repair
275
276 // Failures lists what could not be examined or could not be repaired.
277 Failures []ReconcileFailure
278 }
279
280 // Reconcile runs one pass: scan proposals/* refs and each space's approved
281 // head, compare against rows and index stamps, repair divergence.
282 //
283 // The read order is load-bearing. Refs are listed for every space *before* any
284 // proposal row is read, so a proposal opened concurrently can only ever look
285 // like "row with no branch" — which the grace window protects — and never like
286 // "branch with no row", which would delete a live agent's work. Reversing the
287 // two reads turns an ordinary concurrent propose into data loss.
288 4 func (s *Service) Reconcile(ctx context.Context) (*ReconcileReport, error) {
289 4 spaces, err := s.ListSpaces(ctx)
290 4 if err != nil {
291 1 return nil, err
292 1 }
293
294 3 rep := &ReconcileReport{}
295 3
296 3 // Pass one: every space's repository, approved head and proposal branches.
297 3 type observed struct {
298 3 space *Space
299 3 head plumbing.Hash
300 3 branches []gitx.Branch
301 3 }
302 3 seen := make([]observed, 0, len(spaces))
303 3 for _, sp := range spaces {
304 3 repo, err := s.openRepo(sp.Ref)
305 3 if err != nil {
306 0 rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
307 0 continue
308 }
309 3 sp.Repo = repo
310 3 head, err := repo.ApprovedHead(ctx)
311 3 if err != nil {
312 0 rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
313 0 continue
314 }
315 3 branches, err := repo.ListProposalBranches(ctx)
316 3 if err != nil {
317 0 rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
318 0 continue
319 }
320 3 seen = append(seen, observed{space: sp, head: head, branches: branches})
321 }
322
323 // Pass two: the rows, read strictly after every ref listing above.
324 3 open, err := s.store.ListProposalsByState(ctx, core.StateOpen, 0)
325 3 if err != nil {
326 0 return nil, fmt.Errorf("service: list open proposals: %w", err)
327 0 }
328 3 openBySpace := make(map[int][]*db.Proposal, len(spaces))
329 5 for _, p := range open {
330 5 openBySpace[p.SpaceID] = append(openBySpace[p.SpaceID], p)
331 5 }
332
333 3 for _, o := range seen {
334 3 rep.Spaces++
335 3 facts, err := s.spaceFacts(ctx, o.space, o.head, o.branches, openBySpace[o.space.ID])
336 3 if err != nil {
337 0 rep.Failures = append(rep.Failures, ReconcileFailure{Space: o.space.Ref, Err: err})
338 0 continue
339 }
340 6 for _, r := range PlanRepairs(facts) {
341 6 if r.Kind == RepairReindex {
342 3 rep.Reindex = append(rep.Reindex, r)
343 3 continue
344 }
345 3 if err := s.applyRepair(ctx, o.space, r); err != nil {
346 0 repair := r
347 0 rep.Failures = append(rep.Failures, ReconcileFailure{
348 0 Space: o.space.Ref, Repair: &repair, Err: err,
349 0 })
350 0 continue
351 }
352 3 rep.Repaired = append(rep.Repaired, r)
353 }
354 }
355 3 return rep, nil
356 }
357
358 // spaceFacts turns one space's refs and rows into the facts PlanRepairs
359 // consumes, resolving the two things only I/O can answer: whether a branch has
360 // merged into the approved head, and whether a branch without an *open* row has
361 // any row at all.
362 func (s *Service) spaceFacts(ctx context.Context, sp *Space, head plumbing.Hash,
363 3 branches []gitx.Branch, openRows []*db.Proposal) (SpaceFacts, error) {
364 3
365 3 facts := SpaceFacts{
366 3 Space: sp.Ref,
367 3 SpaceID: sp.ID,
368 3 ApprovedHead: head.String(),
369 3 Now: s.now(),
370 3 Grace: s.grace,
371 3 }
372 3
373 3 stamp, err := s.store.GetIndexStamp(ctx, sp.ID)
374 3 switch {
375 0 case err == nil:
376 0 facts.IndexRev = stamp.Rev
377 case errors.Is(err, db.ErrNotFound):
378 // Never indexed. Left empty, which PlanRepairs reads as stale.
379 0 default:
380 0 return SpaceFacts{}, fmt.Errorf("service: read index stamp for %s: %w", sp.Ref, err)
381 }
382
383 3 byBranch := make(map[string]gitx.Branch, len(branches))
384 5 for _, b := range branches {
385 5 byBranch[b.Name] = b
386 5 }
387
388 3 rowBranches := make(map[string]bool, len(openRows))
389 5 for _, row := range openRows {
390 5 rowBranches[row.Branch] = true
391 5 fact := ProposalFact{
392 5 ID: row.ID,
393 5 Branch: row.Branch,
394 5 HasRow: true,
395 5 State: row.State,
396 5 Created: row.Created,
397 5 }
398 5 if b, ok := byBranch[row.Branch]; ok {
399 3 fact.HasBranch = true
400 3 fact.BranchHead = b.Head.String()
401 3 merged, err := sp.Repo.IsAncestor(ctx, b.Head, head)
402 3 if err != nil {
403 0 return SpaceFacts{}, fmt.Errorf("service: ancestry of %s in %s: %w", row.Branch, sp.Ref, err)
404 0 }
405 3 fact.MergedIntoApproved = merged
406 3 // Resolved rather than compared as a string: base_rev is whatever
407 3 // the agent sent as If-Match, and an abbreviated spelling of the
408 3 // branch tip would otherwise read as "the branch has commits".
409 3 //
410 3 // A base that is no longer in the repository leaves this empty,
411 3 // which PlanRepairs reads as "do not repair" — the row stays open
412 3 // where a human can see it. Any other failure is a real read error
413 3 // and is surfaced rather than silently disarming the check.
414 3 base, err := sp.Repo.ResolveRev(ctx, row.BaseRev)
415 3 switch {
416 3 case err == nil:
417 3 fact.BaseRev = base.String()
418 case errors.Is(err, gitx.ErrNotFound), errors.Is(err, gitx.ErrBadRev):
419 0 default:
420 0 return SpaceFacts{}, fmt.Errorf("service: resolve base %q of %s in %s: %w",
421 0 row.BaseRev, row.Branch, sp.Ref, err)
422 }
423 }
424 5 facts.Proposals = append(facts.Proposals, fact)
425 }
426
427 5 for _, b := range branches {
428 5 if rowBranches[b.Name] {
429 3 continue
430 }
431 // No *open* row claims this branch. A merged or rejected proposal keeps
432 // its branch and its row, so before calling the ref an orphan we ask
433 // whether any row owns it. Getting this wrong deletes the branch of an
434 // already-merged proposal.
435 2 id, ok := gitx.ParseProposalBranch(b.Name)
436 2 if ok {
437 2 row, err := s.store.GetProposal(ctx, int(id))
438 2 switch {
439 1 case err == nil && row.SpaceID == sp.ID && row.Branch == b.Name:
440 1 facts.Proposals = append(facts.Proposals, ProposalFact{
441 1 ID: row.ID, Branch: b.Name, HasRow: true, HasBranch: true,
442 1 State: row.State, Created: row.Created, BranchHead: b.Head.String(),
443 1 })
444 1 continue
445 case err == nil, errors.Is(err, db.ErrNotFound):
446 // Either no row at all, or a row that belongs to another space
447 // or another branch — both mean this ref is unreferenced.
448 0 default:
449 0 return SpaceFacts{}, fmt.Errorf("service: look up proposal %d for %s: %w", id, sp.Ref, err)
450 }
451 }
452 1 facts.Proposals = append(facts.Proposals, ProposalFact{
453 1 ID: int(idOrZero(b.Name)), Branch: b.Name, HasBranch: true, BranchHead: b.Head.String(),
454 1 })
455 }
456 3 return facts, nil
457 }
458
459 1 func idOrZero(branch string) int64 {
460 1 id, ok := gitx.ParseProposalBranch(branch)
461 1 if !ok {
462 0 return 0
463 0 }
464 1 return id
465 }
466
467 // applyRepair executes one repair. RepairReindex never reaches it — Phase 2
468 // owns the index — and an unknown kind is an error rather than a no-op, so a
469 // repair added to PlanRepairs without an implementation fails loudly.
470 4 func (s *Service) applyRepair(ctx context.Context, sp *Space, r Repair) error {
471 4 switch r.Kind {
472 1 case RepairMarkMerged:
473 1 return s.store.MarkProposalMerged(ctx, r.ProposalID, r.Approval, r.Rev)
474 1 case RepairDeleteRow:
475 1 return s.store.DeleteOpenProposal(ctx, r.ProposalID)
476 1 case RepairDeleteRef:
477 1 return s.deleteProposalRef(ctx, sp, r.Branch)
478 1 default:
479 1 return fmt.Errorf("service: no implementation for repair %q", r.Kind)
480 }
481 }
482
483 // deleteProposalRef removes an unreferenced proposals/* branch.
484 //
485 // The namespace check, the per-space write lock and the choice to treat an
486 // already-absent branch as success all live in gitx.DeleteProposalBranch, which
487 // owns refs. What is left here is naming the space the failure belongs to, so
488 // the reconcile report says which repository could not be repaired.
489 3 func (s *Service) deleteProposalRef(ctx context.Context, sp *Space, branch string) error {
490 3 if err := sp.Repo.DeleteProposalBranch(ctx, branch); err != nil {
491 1 return fmt.Errorf("service: delete branch %q in %s: %w", branch, sp.Ref, err)
492 1 }
493 2 return nil
494 }
495
496 // RunReconciler runs the reconciler at startup and then on a ticker, until ctx
497 // is cancelled. report is called with the outcome of every pass; a nil report
498 // callback discards it.
499 //
500 // Running at startup is the half that matters: a daemon killed mid-merge
501 // repairs itself on the next boot with no manual intervention. The ticker
502 // catches the rest — a crash that leaves the daemon running, or a repair that
503 // failed once and succeeds later.
504 1 func (s *Service) RunReconciler(ctx context.Context, interval time.Duration, report func(*ReconcileReport, error)) {
505 1 if interval <= 0 {
506 0 interval = DefaultReconcileInterval
507 0 }
508 1 run := func() {
509 1 rep, err := s.Reconcile(ctx)
510 1 if report != nil {
511 1 report(rep, err)
512 1 }
513 }
514 1 run()
515 1
516 1 ticker := time.NewTicker(interval)
517 1 defer ticker.Stop()
518 1 for {
519 1 select {
520 1 case <-ctx.Done():
521 1 return
522 0 case <-ticker.C:
523 0 run()
524 }
525 }
526 }