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()
}
}
}