~bigbes/sr-ht-spec

ref: 8255ff90741113fd85e6f15706127bb5c30ca3f5 sr-ht-spec/service/reconcile.go -rw-r--r-- 18.3 KiB
8255ff90 — Eugene Blikh ci: export the version instead of sed-ing a tracked APKBUILD 9 days ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
package service

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/go-git/go-git/v5/plumbing"

	"sourcecraft.dev/bigbes/sr-ht-spec/core"
	"sourcecraft.dev/bigbes/sr-ht-spec/db"
	"sourcecraft.dev/bigbes/sr-ht-spec/gitx"
)

// Three systems are touched by a merge — git refs, the bleve index and
// Postgres — and none of it is transactional. The rule that makes it tractable:
//
//	Git refs are the source of truth for whether a proposal has merged. The
//	Postgres row is the source of truth for that a proposal exists and what it
//	is. The index and the render cache are pure caches.
//
// The reconciler is the backstop that repairs divergence, and it implements
// exactly four repairs — no more, because every additional guess about what a
// half-finished write meant is a way to invent state nobody wrote.

const (
	// DefaultReconcileInterval is how often the reconciler runs after startup.
	DefaultReconcileInterval = 15 * time.Minute

	// DefaultReconcileGrace is how long a proposal row with no branch is left
	// alone before it is deleted.
	//
	// The design's repair table has no grace period, and without one the
	// reconciler is actively destructive during ordinary operation: a proposal
	// is opened row-first, so every live propose passes through the exact state
	// ("open row, no branch") that the table says to delete. The window is
	// milliseconds wide and the reconciler runs on a timer, so it would be rare
	// — which makes it worse, not better, since it would destroy an agent's
	// work at random and never in a test.
	//
	// A row younger than this is therefore assumed to be in flight rather than
	// abandoned. It costs one extra reconcile cycle before a genuinely crashed
	// proposal is cleaned up, which nothing is waiting on.
	DefaultReconcileGrace = 5 * time.Minute
)

// RepairKind names one of the four repairs.
type RepairKind string

const (
	// RepairDeleteRow removes an `open` proposal row whose branch does not
	// exist: the daemon died between the row insert and the branch write. The
	// row holds no content, and the agent still holds the document it wanted to
	// write, so it re-proposes.
	RepairDeleteRow RepairKind = "delete-proposal-row"

	// RepairDeleteRef removes a proposals/* ref with no row. It is unreferenced
	// — the id is a Postgres serial, and title, rationale, base_rev, agent and
	// agent_session live nowhere in a ref — so its content is unrecoverable
	// anyway and recreating the row would mean inventing every field.
	RepairDeleteRef RepairKind = "delete-proposal-ref"

	// RepairMarkMerged transitions a row still `open` whose branch has merged
	// into the approved head. The ref is truth for merged-ness.
	RepairMarkMerged RepairKind = "mark-proposal-merged"

	// RepairReindex flags a space whose index stamp differs from its approved
	// head. Phase 2 owns the rebuild; the reconciler only reports the list.
	RepairReindex RepairKind = "reindex-space"
)

// Repair is one repair the reconciler decided on.
type Repair struct {
	Kind    RepairKind
	Space   core.SpaceRef
	SpaceID int

	// ProposalID is the proposal row's id, and zero for RepairReindex and for
	// an orphan ref whose name carries no parseable id.
	ProposalID int

	// Branch is the proposal branch this repair is about, empty for
	// RepairReindex.
	Branch string

	// Rev is the revision the repair records: the merge revision for
	// RepairMarkMerged, the approved head for RepairReindex.
	Rev string

	// Approval is the approval kind RepairMarkMerged records. See
	// PlanRepairs for why it is always core.ApprovalPolicy.
	Approval core.Approval

	// Reason is a human-readable sentence for the log.
	Reason string
}

func (r Repair) String() string {
	return fmt.Sprintf("%s %s: %s", r.Kind, r.Space, r.Reason)
}

// ProposalFact is what the reconciler observed about one proposal — its row,
// its branch, or both. Facts are gathered by I/O and consumed by PlanRepairs,
// which is pure so that the decision table can be exhaustively tested without a
// repository or a database.
type ProposalFact struct {
	// ID is the proposal row id, and the id parsed out of the branch name when
	// there is no row. Zero when the branch name carries no parseable id.
	ID int

	// Branch is the proposal branch name, "proposals/42".
	Branch string

	// HasRow and HasBranch record which halves exist. Both false is not a fact.
	HasRow    bool
	HasBranch bool

	// State is the row's state, meaningless when HasRow is false.
	State core.ProposalState

	// Created is when the row was inserted, for the grace window.
	Created time.Time

	// MergedIntoApproved reports whether the branch tip is reachable from the
	// approved head. Computed by I/O (it needs the object database) and passed
	// in, exactly as gitx.RefUpdate.FastForward is.
	MergedIntoApproved bool

	// BranchHead is the branch tip.
	BranchHead string

	// BaseRev is the proposal's recorded base, resolved to an object name, and
	// empty when it could not be resolved. See PlanRepairs for why a branch
	// still sitting on its base is not a merge.
	BaseRev string
}

// SpaceFacts is everything the reconciler observed about one space.
type SpaceFacts struct {
	Space   core.SpaceRef
	SpaceID int

	// ApprovedHead is the current tip of the approved branch.
	ApprovedHead string

	// IndexRev is the revision the global index currently reflects for this
	// space, empty when the space has never been indexed. Empty is stale by
	// construction, which is why a missing stamp is not treated as up to date.
	IndexRev string

	Proposals []ProposalFact

	// Now and Grace parameterize the grace window, so it is an input to the
	// decision rather than a clock read inside it.
	Now   time.Time
	Grace time.Duration
}

// PlanRepairs is the repair table, as a pure function.
//
// It implements exactly the four rows of the design's "Consistency and
// recovery" table and nothing else. States it does not name — a merged row
// whose branch is no longer an ancestor of the approved head, a rejected row
// whose branch still exists — are deliberately left alone: neither is a
// half-finished write, and repairing them would mean deciding something the
// design did not.
//
// "Merged" needs one qualification the design's table does not state, and
// without it the reconciler corrupts state during ordinary operation. A
// proposal branch is cut *at* the approved head, so between the cut and the
// agent's first commit its tip is trivially an ancestor of that head — and the
// literal rule "branch merged into the approved head, row still open" fires on
// a proposal that has not merged and has no content at all. The same holds
// forever after for a proposal whose agent never committed. A branch still
// sitting on its recorded base is therefore never treated as merged, and a base
// that could not be resolved is treated the same way: repairing on facts we
// could not establish is worse than leaving the row open for a human to see.
//
// RepairMarkMerged always records core.ApprovalPolicy. The ref proves the
// merge happened and nothing proves how it was authorized — the approval kind
// existed only in the memory of the process that died. Of the two available
// lies, "policy" is the safe one: recording "human" would launder unreviewed
// content as blessed, which is the exact failure the bimodal decision exists to
// prevent, while recording "policy" understates the review and puts the
// proposal in the policy-merged digest, where a human sees it again. Erring
// toward visibility is the whole point of the digest.
func PlanRepairs(f SpaceFacts) []Repair {
	var repairs []Repair
	base := Repair{Space: f.Space, SpaceID: f.SpaceID}

	for _, p := range f.Proposals {
		r := base
		r.ProposalID = p.ID
		r.Branch = p.Branch

		switch {
		case p.HasRow && p.HasBranch && p.State == core.StateOpen && p.MergedIntoApproved &&
			p.BaseRev != "" && p.BranchHead != p.BaseRev:
			r.Kind = RepairMarkMerged
			r.Rev = f.ApprovedHead
			r.Approval = core.ApprovalPolicy
			r.Reason = fmt.Sprintf("branch %s has merged into the approved head %s but the row is still open",
				p.Branch, short(f.ApprovedHead))
			repairs = append(repairs, r)

		case p.HasRow && !p.HasBranch && p.State == core.StateOpen:
			if f.Now.Sub(p.Created) < f.Grace {
				continue // in flight: the row is written before the branch
			}
			r.Kind = RepairDeleteRow
			r.Reason = fmt.Sprintf("row is open but branch %s does not exist; the agent re-proposes", p.Branch)
			repairs = append(repairs, r)

		case !p.HasRow && p.HasBranch:
			r.Kind = RepairDeleteRef
			r.Reason = fmt.Sprintf("branch %s has no row; its content is unrecoverable", p.Branch)
			repairs = append(repairs, r)
		}
	}

	if f.ApprovedHead != "" && f.IndexRev != f.ApprovedHead {
		r := base
		r.Kind = RepairReindex
		r.Rev = f.ApprovedHead
		r.Reason = fmt.Sprintf("index stamp %s differs from the approved head %s",
			stampOrNever(f.IndexRev), short(f.ApprovedHead))
		repairs = append(repairs, r)
	}
	return repairs
}

func short(rev string) string {
	if len(rev) > 8 {
		return rev[:8]
	}
	return rev
}

func stampOrNever(rev string) string {
	if rev == "" {
		return "(never indexed)"
	}
	return short(rev)
}

// ReconcileFailure is one thing the reconciler could not do. Failures never
// abort the run: a space with an unreadable repository must not stop the other
// spaces from being repaired.
type ReconcileFailure struct {
	Space  core.SpaceRef
	Repair *Repair // nil when the whole space could not be examined
	Err    error
}

func (f ReconcileFailure) Error() string {
	if f.Repair != nil {
		return fmt.Sprintf("%s: %v", f.Repair, f.Err)
	}
	return fmt.Sprintf("%s: %v", f.Space, f.Err)
}

// ReconcileReport is what one reconciler pass did.
type ReconcileReport struct {
	// Spaces is how many spaces were examined.
	Spaces int

	// Repaired lists the repairs that were applied.
	Repaired []Repair

	// Reindex lists spaces whose index is stale. They are reported, not
	// repaired: bleve is single-writer and Phase 2 owns the index. A caller
	// that has an indexer drives it from this list.
	Reindex []Repair

	// Failures lists what could not be examined or could not be repaired.
	Failures []ReconcileFailure
}

// Reconcile runs one pass: scan proposals/* refs and each space's approved
// head, compare against rows and index stamps, repair divergence.
//
// The read order is load-bearing. Refs are listed for every space *before* any
// proposal row is read, so a proposal opened concurrently can only ever look
// like "row with no branch" — which the grace window protects — and never like
// "branch with no row", which would delete a live agent's work. Reversing the
// two reads turns an ordinary concurrent propose into data loss.
func (s *Service) Reconcile(ctx context.Context) (*ReconcileReport, error) {
	spaces, err := s.ListSpaces(ctx)
	if err != nil {
		return nil, err
	}

	rep := &ReconcileReport{}

	// Pass one: every space's repository, approved head and proposal branches.
	type observed struct {
		space    *Space
		head     plumbing.Hash
		branches []gitx.Branch
	}
	seen := make([]observed, 0, len(spaces))
	for _, sp := range spaces {
		repo, err := s.openRepo(sp.Ref)
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
			continue
		}
		sp.Repo = repo
		head, err := repo.ApprovedHead(ctx)
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
			continue
		}
		branches, err := repo.ListProposalBranches(ctx)
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err})
			continue
		}
		seen = append(seen, observed{space: sp, head: head, branches: branches})
	}

	// Pass two: the rows, read strictly after every ref listing above.
	open, err := s.store.ListProposalsByState(ctx, core.StateOpen, 0)
	if err != nil {
		return nil, fmt.Errorf("service: list open proposals: %w", err)
	}
	openBySpace := make(map[int][]*db.Proposal, len(spaces))
	for _, p := range open {
		openBySpace[p.SpaceID] = append(openBySpace[p.SpaceID], p)
	}

	for _, o := range seen {
		rep.Spaces++
		facts, err := s.spaceFacts(ctx, o.space, o.head, o.branches, openBySpace[o.space.ID])
		if err != nil {
			rep.Failures = append(rep.Failures, ReconcileFailure{Space: o.space.Ref, Err: err})
			continue
		}
		for _, r := range PlanRepairs(facts) {
			if r.Kind == RepairReindex {
				rep.Reindex = append(rep.Reindex, r)
				continue
			}
			if err := s.applyRepair(ctx, o.space, r); err != nil {
				repair := r
				rep.Failures = append(rep.Failures, ReconcileFailure{
					Space: o.space.Ref, Repair: &repair, Err: err,
				})
				continue
			}
			rep.Repaired = append(rep.Repaired, r)
		}
	}
	return rep, nil
}

// spaceFacts turns one space's refs and rows into the facts PlanRepairs
// consumes, resolving the two things only I/O can answer: whether a branch has
// merged into the approved head, and whether a branch without an *open* row has
// any row at all.
func (s *Service) spaceFacts(ctx context.Context, sp *Space, head plumbing.Hash,
	branches []gitx.Branch, openRows []*db.Proposal) (SpaceFacts, error) {

	facts := SpaceFacts{
		Space:        sp.Ref,
		SpaceID:      sp.ID,
		ApprovedHead: head.String(),
		Now:          s.now(),
		Grace:        s.grace,
	}

	stamp, err := s.store.GetIndexStamp(ctx, sp.ID)
	switch {
	case err == nil:
		facts.IndexRev = stamp.Rev
	case errors.Is(err, db.ErrNotFound):
		// Never indexed. Left empty, which PlanRepairs reads as stale.
	default:
		return SpaceFacts{}, fmt.Errorf("service: read index stamp for %s: %w", sp.Ref, err)
	}

	byBranch := make(map[string]gitx.Branch, len(branches))
	for _, b := range branches {
		byBranch[b.Name] = b
	}

	rowBranches := make(map[string]bool, len(openRows))
	for _, row := range openRows {
		rowBranches[row.Branch] = true
		fact := ProposalFact{
			ID:      row.ID,
			Branch:  row.Branch,
			HasRow:  true,
			State:   row.State,
			Created: row.Created,
		}
		if b, ok := byBranch[row.Branch]; ok {
			fact.HasBranch = true
			fact.BranchHead = b.Head.String()
			merged, err := sp.Repo.IsAncestor(ctx, b.Head, head)
			if err != nil {
				return SpaceFacts{}, fmt.Errorf("service: ancestry of %s in %s: %w", row.Branch, sp.Ref, err)
			}
			fact.MergedIntoApproved = merged
			// Resolved rather than compared as a string: base_rev is whatever
			// the agent sent as If-Match, and an abbreviated spelling of the
			// branch tip would otherwise read as "the branch has commits".
			//
			// A base that is no longer in the repository leaves this empty,
			// which PlanRepairs reads as "do not repair" — the row stays open
			// where a human can see it. Any other failure is a real read error
			// and is surfaced rather than silently disarming the check.
			base, err := sp.Repo.ResolveRev(ctx, row.BaseRev)
			switch {
			case err == nil:
				fact.BaseRev = base.String()
			case errors.Is(err, gitx.ErrNotFound), errors.Is(err, gitx.ErrBadRev):
			default:
				return SpaceFacts{}, fmt.Errorf("service: resolve base %q of %s in %s: %w",
					row.BaseRev, row.Branch, sp.Ref, err)
			}
		}
		facts.Proposals = append(facts.Proposals, fact)
	}

	for _, b := range branches {
		if rowBranches[b.Name] {
			continue
		}
		// No *open* row claims this branch. A merged or rejected proposal keeps
		// its branch and its row, so before calling the ref an orphan we ask
		// whether any row owns it. Getting this wrong deletes the branch of an
		// already-merged proposal.
		id, ok := gitx.ParseProposalBranch(b.Name)
		if ok {
			row, err := s.store.GetProposal(ctx, int(id))
			switch {
			case err == nil && row.SpaceID == sp.ID && row.Branch == b.Name:
				facts.Proposals = append(facts.Proposals, ProposalFact{
					ID: row.ID, Branch: b.Name, HasRow: true, HasBranch: true,
					State: row.State, Created: row.Created, BranchHead: b.Head.String(),
				})
				continue
			case err == nil, errors.Is(err, db.ErrNotFound):
				// Either no row at all, or a row that belongs to another space
				// or another branch — both mean this ref is unreferenced.
			default:
				return SpaceFacts{}, fmt.Errorf("service: look up proposal %d for %s: %w", id, sp.Ref, err)
			}
		}
		facts.Proposals = append(facts.Proposals, ProposalFact{
			ID: int(idOrZero(b.Name)), Branch: b.Name, HasBranch: true, BranchHead: b.Head.String(),
		})
	}
	return facts, nil
}

func idOrZero(branch string) int64 {
	id, ok := gitx.ParseProposalBranch(branch)
	if !ok {
		return 0
	}
	return id
}

// applyRepair executes one repair. RepairReindex never reaches it — Phase 2
// owns the index — and an unknown kind is an error rather than a no-op, so a
// repair added to PlanRepairs without an implementation fails loudly.
func (s *Service) applyRepair(ctx context.Context, sp *Space, r Repair) error {
	switch r.Kind {
	case RepairMarkMerged:
		return s.store.MarkProposalMerged(ctx, r.ProposalID, r.Approval, r.Rev)
	case RepairDeleteRow:
		return s.store.DeleteOpenProposal(ctx, r.ProposalID)
	case RepairDeleteRef:
		return s.deleteProposalRef(ctx, sp, r.Branch)
	default:
		return fmt.Errorf("service: no implementation for repair %q", r.Kind)
	}
}

// deleteProposalRef removes an unreferenced proposals/* branch.
//
// The namespace check, the per-space write lock and the choice to treat an
// already-absent branch as success all live in gitx.DeleteProposalBranch, which
// owns refs. What is left here is naming the space the failure belongs to, so
// the reconcile report says which repository could not be repaired.
func (s *Service) deleteProposalRef(ctx context.Context, sp *Space, branch string) error {
	if err := sp.Repo.DeleteProposalBranch(ctx, branch); err != nil {
		return fmt.Errorf("service: delete branch %q in %s: %w", branch, sp.Ref, err)
	}
	return nil
}

// RunReconciler runs the reconciler at startup and then on a ticker, until ctx
// is cancelled. report is called with the outcome of every pass; a nil report
// callback discards it.
//
// Running at startup is the half that matters: a daemon killed mid-merge
// repairs itself on the next boot with no manual intervention. The ticker
// catches the rest — a crash that leaves the daemon running, or a repair that
// failed once and succeeds later.
func (s *Service) RunReconciler(ctx context.Context, interval time.Duration, report func(*ReconcileReport, error)) {
	if interval <= 0 {
		interval = DefaultReconcileInterval
	}
	run := func() {
		rep, err := s.Reconcile(ctx)
		if report != nil {
			report(rep, err)
		}
	}
	run()

	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			run()
		}
	}
}