From 275ddd76e88f539ca97ae803e7752b04fa1983d5 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 03:06:50 +0200 Subject: [PATCH] Keep rollouts and the merges they answer in the rollouts bucket (issue 1092) Once chain_moved records the kind, a plan is a key of the rollouts bucket under its id and the open batch is the same key in state assembling. A save refuses a copy another save moved since (the plan's own Revision, kept in its value, counts saves) and keeps the merges the record holds, so a copy read before a merge moved cannot lay it back. The merges a batch or walk answers are the record's held field, with the columns the table had; a merge no plan answers stays, flagged, in the record that held it last, or in the record kept for those. A merge moved to a plan is written to the receiving record first and then taken from the giving one; one found in both is read from the one it entered last and repaired at the next cut. Plans and merges are read from a view kept by one watch of the bucket. A bus that does not answer, a lease lost and a key moved are now not-now like the store being away, so a merge whose plan could not be written is held and handed over again, never acknowledged. The start moves release_plan and batched_merge into the bucket, one kind, counted against both tables. --- cmd/mesh-controller/batches.go | 12 +- .../merge_unsaved_integration_test.go | 71 ++ cmd/mesh-controller/merge_unsaved_test.go | 26 + cmd/mesh-controller/upgrades.go | 10 + internal/chain/backend.go | 75 +- internal/chain/chain.go | 40 +- internal/chain/records.go | 16 + internal/inventory/batches.go | 40 + internal/inventory/chain.go | 15 +- internal/inventory/chain_builds_test.go | 31 + internal/inventory/chain_move.go | 89 ++ internal/inventory/chain_move_test.go | 101 ++ internal/inventory/chain_rollouts.go | 939 ++++++++++++++++++ internal/inventory/chain_rollouts_test.go | 352 +++++++ internal/inventory/fortest.go | 12 +- internal/inventory/nodes.go | 15 +- internal/inventory/plan_id_test.go | 16 +- internal/inventory/plans.go | 45 + internal/inventory/reachable.go | 12 + internal/link/source_moved_held_test.go | 60 ++ testdata/store-tests | 2 + 21 files changed, 1939 insertions(+), 40 deletions(-) create mode 100644 cmd/mesh-controller/merge_unsaved_integration_test.go create mode 100644 cmd/mesh-controller/merge_unsaved_test.go create mode 100644 internal/inventory/chain_rollouts.go create mode 100644 internal/inventory/chain_rollouts_test.go create mode 100644 internal/link/source_moved_held_test.go diff --git a/cmd/mesh-controller/batches.go b/cmd/mesh-controller/batches.go index b3c09aff..1acbe182 100644 --- a/cmd/mesh-controller/batches.go +++ b/cmd/mesh-controller/batches.go @@ -219,10 +219,10 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T } if p.Batch() { if err := keepBatch(ctx, inv, &p, now, ""); err != nil { - return notNow(err) + return unsaved(err) } } else if err := nameMerges(ctx, inv, &p); err != nil { - return notNow(err) + return unsaved(err) } fmt.Printf("%s merged into %s (%.8s) after a later merge of its branch was put in %s (%s): that %s "+ "answers it\n", repository, m.Base, m.Commit, p.ID, p.State, walkOrBatch(p)) @@ -239,7 +239,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T } holdFor(&batch, m.Head, waits, holding, own, now) if err := keepBatch(ctx, inv, &batch, now, ""); err != nil { - return notNow(err) + return unsaved(err) } fmt.Printf("%s merged into %s (%.8s); %s, it joins %s — %s\n", repository, m.Base, m.Commit, why, batch.ID, batchWords(batch, now)) @@ -820,6 +820,12 @@ func nameMerges(ctx context.Context, inv *inventory.Inventory, p *inventory.Plan // walk open, behind which it is said queued (cutUnrelated). While the search runs everything waits. func cutBatchesHeld(ctx context.Context, open *stores, now time.Time) error { inv := open.inventory + // A merge found in two records is the half-done move of a crash: put right before anything is decided from it. + if repaired, err := inv.RepairMerges(ctx); err != nil { + return err + } else if repaired > 0 { + fmt.Printf("%d merge(s) found in two plans, the half of a move a restart interrupted, are kept by the plan they were moved to\n", repaired) + } if err := searchAfterFailures(ctx, inv); err != nil { return err } diff --git a/cmd/mesh-controller/merge_unsaved_integration_test.go b/cmd/mesh-controller/merge_unsaved_integration_test.go new file mode 100644 index 00000000..095930b9 --- /dev/null +++ b/cmd/mesh-controller/merge_unsaved_integration_test.go @@ -0,0 +1,71 @@ +//go:build integration + +package main + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/link" +) + +// refusing is a bucket that refuses to write while it is told to. +type refusing struct { + chain.Backend + refuse atomic.Bool +} + +func (r *refusing) Create(ctx context.Context, key string, value []byte) (uint64, error) { + if r.refuse.Load() { + return 0, nats.ErrTimeout + } + return r.Backend.Create(ctx, key, value) +} + +func (r *refusing) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + if r.refuse.Load() { + return 0, nats.ErrTimeout + } + return r.Backend.Update(ctx, key, value, revision) +} + +// **A merge whose plan cannot be saved is held, recorded nowhere, and recorded when it is handed over again** (ADR 0407 +// decision 3, issue 1092): hearMerge answers ErrTryAgain — the bus leaves the announcement unacknowledged and hands it +// over again — and the merge is in no record until the plan holding it is written. +func TestAMergeWhosePlanCannotBeSavedIsHeldAndRecordedWhenHandedOverAgain(t *testing.T) { + open := windowed(t) + everyHeadWaits(t) + asksWithPaths(t) + ctx := t.Context() + rollouts := &refusing{Backend: chain.NewMemory()} + open.inventory.UseChain(chain.New(chain.NewMemory(), rollouts, chain.NewMemory(), func(context.Context) error { return nil })) + merge := catalogueMerge("cat1aaaaaaaa", "app", t0) + + rollouts.refuse.Store(true) + err := (following{open: open}).hearMerge(ctx, merge, t0.Add(time.Second)) + if !errors.Is(err, link.ErrTryAgain) { + t.Fatalf("a merge whose plan could not be saved answered %v, which the bus takes as done", err) + } + if _, known, err := open.inventory.MergeOf(ctx, "novox/mesh-catalog", merge.Commit); err != nil || known { + t.Fatalf("the merge is kept though its plan was not: %v %v", known, err) + } + if _, bs := walks(t, open); len(bs) != 0 { + t.Fatalf("a batch was kept though its write failed: %+v", bs) + } + + rollouts.refuse.Store(false) + hear(t, open, merge, t0.Add(10*time.Second)) + _, bs := walks(t, open) + if len(bs) != 1 { + t.Fatalf("the merge handed over again made %d batch(es)", len(bs)) + } + if m, known, _ := open.inventory.MergeOf(ctx, "novox/mesh-catalog", merge.Commit); !known || m.Plan != bs[0].ID { + t.Fatalf("the merge is %+v %v, and the batch %s", m, known, bs[0].ID) + } +} diff --git a/cmd/mesh-controller/merge_unsaved_test.go b/cmd/mesh-controller/merge_unsaved_test.go new file mode 100644 index 00000000..c967060f --- /dev/null +++ b/cmd/mesh-controller/merge_unsaved_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "errors" + "testing" + + "github.com/novox/mesh-controller/internal/link" +) + +// **A merge whose plan could not be saved is held, never taken** (novox/hq ADR 0407 decision 3, issue 1092): the merge is +// carried until its plan is saved, and the bus acknowledges the announcement only when the controller answers without +// ErrTryAgain (link.Server.sourceMoved), so a failed save of any kind must answer it. +func TestAFailureToSaveAMergesPlanIsHeldAndNotTaken(t *testing.T) { + cause := errors.New("the plan for novox/mesh-catalog is not written: the plan was written by somebody else since it was read") + held := unsaved(cause) + if !errors.Is(held, link.ErrTryAgain) || !errors.Is(held, cause) { + t.Fatalf("a failed save answers %v", held) + } + if unsaved(nil) != nil { + t.Fatal("a save that worked answers an error") + } + // Not the same as notNow, which holds a merge only for a store that is away. + if errors.Is(notNow(cause), link.ErrTryAgain) { + t.Fatal("notNow holds every failure; the premise of this test is gone") + } +} diff --git a/cmd/mesh-controller/upgrades.go b/cmd/mesh-controller/upgrades.go index ea9ec498..94d25d4d 100644 --- a/cmd/mesh-controller/upgrades.go +++ b/cmd/mesh-controller/upgrades.go @@ -303,6 +303,16 @@ func notNow(err error) error { return err } +// unsaved marks the failure to save the plan a merge was added to, whatever it was: the merge is carried until that save +// writes it (inventory.AddMerge), so the announcement is held and handed over again, never taken (novox/hq ADR 0407 +// decision 3, ADR 0408, issue 1092). The bus acknowledges it only after the save succeeded. +func unsaved(err error) error { + if err == nil { + return nil + } + return fmt.Errorf("%w: the merge is not kept until its plan is saved: %w", link.ErrTryAgain, err) +} + // SourceMoved is the forge announcing a merge. It is put into the open batch (novox/hq ADR 0276), which // becomes one walk when its merge window closes: hearMerge, and cutBatchesHeld in batches.go. func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error { diff --git a/internal/chain/backend.go b/internal/chain/backend.go index caf52128..486095c7 100644 --- a/internal/chain/backend.go +++ b/internal/chain/backend.go @@ -55,7 +55,7 @@ type Memory struct { mu sync.Mutex values map[string][]Entry revision uint64 - watchers []chan Entry + watchers []*watcher // Now is the clock the entries are dated by; nil is time.Now. Now func() time.Time // writes counts the creates and updates taken. @@ -144,10 +144,7 @@ func (m *Memory) write(key string, value []byte) uint64 { } m.values[key] = h for _, w := range m.watchers { - select { - case w <- clone(e): - default: - } + w.push(clone(e)) } return e.Revision } @@ -166,14 +163,41 @@ func (m *Memory) Keys(_ context.Context) ([]string, error) { return out, nil } +// watcher is one watch of a Memory: an unbounded queue, so a watch that is slow is behind and never misses a write. +type watcher struct { + mu sync.Mutex + q []Entry + wake chan struct{} +} + +func (w *watcher) push(e Entry) { + w.mu.Lock() + w.q = append(w.q, e) + w.mu.Unlock() + select { + case w.wake <- struct{}{}: + default: + } +} + +func (w *watcher) pop() (Entry, bool) { + w.mu.Lock() + defer w.mu.Unlock() + if len(w.q) == 0 { + return Entry{}, false + } + e := w.q[0] + w.q = w.q[1:] + return e, true +} + func (m *Memory) Watch(ctx context.Context) (<-chan Entry, error) { m.mu.Lock() defer m.mu.Unlock() if m.fail != nil { return nil, m.fail } - // Room for the current values, the caught mark, and a burst of later writes. - w := make(chan Entry, len(m.values)+1+1024) + w := &watcher{wake: make(chan struct{}, 1)} keys := make([]string, 0, len(m.values)) for k := range m.values { keys = append(keys, k) @@ -181,30 +205,37 @@ func (m *Memory) Watch(ctx context.Context) (<-chan Entry, error) { sort.Strings(keys) for _, k := range keys { h := m.values[k] - w <- clone(h[len(h)-1]) + w.q = append(w.q, clone(h[len(h)-1])) } - w <- Entry{Caught: true} + w.q = append(w.q, Entry{Caught: true}) m.watchers = append(m.watchers, w) out := make(chan Entry) go func() { defer close(out) + defer func() { + m.mu.Lock() + for i, x := range m.watchers { + if x == w { + m.watchers = append(m.watchers[:i], m.watchers[i+1:]...) + break + } + } + m.mu.Unlock() + }() for { - select { - case <-ctx.Done(): - m.mu.Lock() - for i, x := range m.watchers { - if x == w { - m.watchers = append(m.watchers[:i], m.watchers[i+1:]...) - break - } - } - m.mu.Unlock() - return - case e := <-w: + e, ok := w.pop() + if !ok { select { - case out <- e: case <-ctx.Done(): + return + case <-w.wake: } + continue + } + select { + case out <- e: + case <-ctx.Done(): + return } } }() diff --git a/internal/chain/chain.go b/internal/chain/chain.go index 0b987229..bb3f6862 100644 --- a/internal/chain/chain.go +++ b/internal/chain/chain.go @@ -35,6 +35,11 @@ var validKey = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`) // ErrBadKey is a key the bus does not take: refused loudly, never rewritten into another. var ErrBadKey = errors.New("the bus takes no key of that spelling") +// ValidKey says the bus takes the key: letters, digits and - _ / = . with no dot at either end. +func ValidKey(key string) bool { + return validKey.MatchString(key) && !strings.HasPrefix(key, ".") && !strings.HasSuffix(key, ".") +} + func (s Bucket[T]) guard(ctx context.Context, key string, value []byte) error { if s.holds == nil { return ErrNotHeld @@ -42,7 +47,7 @@ func (s Bucket[T]) guard(ctx context.Context, key string, value []byte) error { if err := s.holds(ctx); err != nil { return fmt.Errorf("%w: %v", ErrNotHeld, err) } - if !validKey.MatchString(key) || strings.HasPrefix(key, ".") || strings.HasSuffix(key, ".") { + if !ValidKey(key) { return fmt.Errorf("%w: %q", ErrBadKey, key) } if len(value) > broker.ChainValueBytes { @@ -147,19 +152,52 @@ func (s Bucket[T]) Keys(ctx context.Context) ([]string, error) { return s.b.Keys // Watch is the bucket's raw entries: its current values, a Caught mark, then each write. func (s Bucket[T]) Watch(ctx context.Context) (<-chan Entry, error) { return s.b.Watch(ctx) } +// GetRaw is one key's value as the bucket holds it, for a writer that changes a field of a record and must keep the rest +// exactly as it is. +func (s Bucket[T]) GetRaw(ctx context.Context, key string) (Entry, bool, error) { + e, ok, err := s.b.Get(ctx, key) + e.Key = key + return e, ok, err +} + +// CreateRaw writes a key's first value as bytes, with the same guards as Create. +func (s Bucket[T]) CreateRaw(ctx context.Context, key string, value []byte) (uint64, error) { + if err := s.guard(ctx, key, value); err != nil { + return 0, err + } + return s.b.Create(ctx, key, value) +} + +// UpdateRaw writes a key's value as bytes, at the revision it was read at, with the same guards as Update. +func (s Bucket[T]) UpdateRaw(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + if err := s.guard(ctx, key, value); err != nil { + return 0, err + } + return s.b.Update(ctx, key, value, revision) +} + // Chain is the three buckets of the chain. type Chain struct { Builds Bucket[Build] Rollouts Bucket[Rollout] Assignments Bucket[Assignment] + + rollouts Backend + holds Holds } +// RolloutsOf is the rollouts bucket typed as the controller's plan, which carries more than the chain's Rollout names: +// the same keys and values, read and written as the plan. +func RolloutsOf[T any](c *Chain) Bucket[T] { return Bucket[T]{c.rollouts, c.holds} } + // New is a chain over three backends, written only while holds says this controller may write. A nil holds writes nothing. func New(builds, rollouts, assignments Backend, holds Holds) *Chain { return &Chain{ Builds: Bucket[Build]{builds, holds}, Rollouts: Bucket[Rollout]{rollouts, holds}, Assignments: Bucket[Assignment]{assignments, holds}, + rollouts: rollouts, + holds: holds, } } diff --git a/internal/chain/records.go b/internal/chain/records.go index aa8ef078..c259e241 100644 --- a/internal/chain/records.go +++ b/internal/chain/records.go @@ -143,6 +143,22 @@ type Links struct { Changed []Change `json:"changed,omitempty"` } +// HeldMerge is a merge a rollout holds: the columns the controller's table of batched merges had, kept in the record of the +// batch or walk that answers it (novox/hq ADR 0276, issue 1092). A merge no plan answers (Waiting) stays in the record +// that held it last. Placed is when it entered this record, so one found in two records after a crash between the two +// writes is read from the one it entered last. +type HeldMerge struct { + Repository string `json:"repository"` + Branch string `json:"branch,omitempty"` + Commit string `json:"commit"` + Merged time.Time `json:"merged,omitzero"` + Heard time.Time `json:"heard,omitzero"` + Event json.RawMessage `json:"event,omitempty"` + Alone bool `json:"alone,omitempty"` + Waiting bool `json:"waiting,omitempty"` + Placed time.Time `json:"placed,omitzero"` +} + // Rollout is the controller's record of one batch's delivery, kept under the batch's id: the commit it fixed per // repository, its tiers, the builds it carried and the assignments it changed (ADR 0408 §1). The controller's plan // carries more (its modules, its times, its backups) beside these, under the same key. diff --git a/internal/inventory/batches.go b/internal/inventory/batches.go index 5e3c5e22..c0cd6ab2 100644 --- a/internal/inventory/batches.go +++ b/internal/inventory/batches.go @@ -33,6 +33,11 @@ type BatchedMerge struct { // AddMerge keeps a merge for a plan; false when the merge was kept already, for whatever plan. func (i *Inventory) AddMerge(ctx context.Context, m BatchedMerge) (bool, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return false, err + } else if on { + return i.chainAddMerge(ctx, c, m) + } tag, err := i.store.Pool().Exec(ctx, `insert into batched_merge (repository, branch, commit_hash, merged_at, heard_at, event, plan_id, alone) values (lower($1), $2, $3, $4, $5, $6, nullif($7, ''), $8) on conflict do nothing`, @@ -45,6 +50,11 @@ func (i *Inventory) AddMerge(ctx context.Context, m BatchedMerge) (bool, error) // MergeOf is one kept merge; false when it was never heard. func (i *Inventory) MergeOf(ctx context.Context, repository, commit string) (BatchedMerge, bool, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return BatchedMerge{}, false, err + } else if on { + return i.chainMergeOf(ctx, c, repository, commit) + } ms, err := i.merges(ctx, `where repository = lower($1) and commit_hash = $2`, repository, commit) if err != nil || len(ms) == 0 { return BatchedMerge{}, false, err @@ -54,12 +64,22 @@ func (i *Inventory) MergeOf(ctx context.Context, repository, commit string) (Bat // MergesOf is every merge a plan answers, oldest merge first. func (i *Inventory) MergesOf(ctx context.Context, plan string) ([]BatchedMerge, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + return i.chainMergesOf(ctx, c, plan) + } return i.merges(ctx, `where plan_id = $1 order by merged_at nulls last, heard_at`, plan) } // LaterMergesOf is every merge of a repository's branch made after a moment and answered by a plan, the // newest first. func (i *Inventory) LaterMergesOf(ctx context.Context, repository, branch string, after time.Time) ([]BatchedMerge, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + return i.chainLaterMergesOf(ctx, c, repository, branch, after) + } return i.merges(ctx, `where repository = lower($1) and branch = $2 and merged_at > $3 and plan_id is not null order by merged_at desc`, repository, branch, after) } @@ -67,6 +87,11 @@ func (i *Inventory) LaterMergesOf(ctx context.Context, repository, branch string // KeptOnBranch says the controller keeps a merge of a repository's branch, and when it heard the branch's first: a merge of a branch it keeps merges of, made since then, is the record's to answer, whatever a // module's last look says (novox/hq ADR 0276). func (i *Inventory) KeptOnBranch(ctx context.Context, repository, branch string) (bool, time.Time, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return false, time.Time{}, err + } else if on { + return i.chainKeptOnBranch(ctx, c, repository, branch) + } var first *time.Time var kept bool err := i.store.Pool().QueryRow(ctx, @@ -80,12 +105,22 @@ func (i *Inventory) KeptOnBranch(ctx context.Context, repository, branch string) // AloneMerges is every merge waiting to be walked on its own commit, the newest first (ADR 0276 decision 3). func (i *Inventory) AloneMerges(ctx context.Context) ([]BatchedMerge, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + return i.chainAloneMerges(ctx, c) + } return i.merges(ctx, `where plan_id is null and alone order by merged_at desc nulls last, heard_at desc`) } // AnswerMerge moves a merge to the plan that answers it now; an empty plan with alone leaves it waiting to // be walked on its own commit. func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan string, alone bool) error { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return err + } else if on { + return i.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: repository, Commit: commit, Plan: plan, Alone: alone}}) + } _, err := i.store.Pool().Exec(ctx, `update batched_merge set plan_id = nullif($3, ''), alone = $4 where repository = lower($1) and commit_hash = $2`, repository, commit, plan, alone) @@ -95,6 +130,11 @@ func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan st // Batches is every batch not yet cut: assembling or queued, oldest first. One at a time is the rule; more // are read so a fault is seen. func (i *Inventory) Batches(ctx context.Context) ([]Plan, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + return i.chainPlans(ctx, c, func(e viewEntry) bool { return e.sum.State == PlanAssembling || e.sum.State == PlanQueued }, sortCreated) + } return i.plans(ctx, `where state in ('assembling', 'queued') order by created`) } diff --git a/internal/inventory/chain.go b/internal/inventory/chain.go index 89fac89a..5b2b42c2 100644 --- a/internal/inventory/chain.go +++ b/internal/inventory/chain.go @@ -19,9 +19,14 @@ import ( // The kinds of the chain, as chain_moved names them. const ( - KindBuilds = "builds" + KindBuilds = "builds" + KindRollouts = "rollouts" ) +// testChains are the chains in memory a test made for its store, by the store's address: every inventory opened on that +// store is given the same chain, as every process of a mesh is on the same bus. Empty outside a test. +var testChains sync.Map + // ErrNoChain is a kind moved to the bus read by an inventory that has no chain attached: a process that did not // open the bus. Refused loudly, never answered from the tables the move left behind. var ErrNoChain = errors.New("this kind of record lives on the bus, and this process has not opened the bus") @@ -32,6 +37,10 @@ type movedKinds struct { mu sync.Mutex kind map[string]bool open ChainOpener + // view is the rollouts bucket kept in memory by one watch (issue 1092); pending are the merges added to a plan that + // is not saved yet. + view *rolloutView + pending map[string]pendingMerge } // ChainOpener opens the chain on the bus when a process first needs it: a command that reads and never moved anything has @@ -51,6 +60,10 @@ func (i *Inventory) UseChain(c *chain.Chain) { i.moved.mu.Lock() defer i.moved.mu.Unlock() i.chain = c + if i.moved.view != nil { + i.moved.view.stop() + i.moved.view = nil + } } // attached is the chain this inventory has, nil when none. diff --git a/internal/inventory/chain_builds_test.go b/internal/inventory/chain_builds_test.go index 8cee9d17..348b20d1 100644 --- a/internal/inventory/chain_builds_test.go +++ b/internal/inventory/chain_builds_test.go @@ -1,11 +1,15 @@ package inventory import ( + "context" "encoding/json" + "fmt" "reflect" "testing" "time" + "github.com/nats-io/nats.go" + "github.com/novox/mesh-controller/internal/chain" ) @@ -98,3 +102,30 @@ func TestBuildsAreOrderedNewestRequestFirst(t *testing.T) { t.Fatalf("ordered %v, not %v", got, want) } } + +// **A bus that does not answer, a lease lost and a key written first are "not now"** (issue 1092): a merge or a report whose +// write failed so is held and handed over again, as it is for a store that is away. +func TestTheBusBeingAwayIsNotNowLikeTheStoreBeingAway(t *testing.T) { + for name, err := range map[string]error{ + "a timeout": fmt.Errorf("the plan is not written: %w", nats.ErrTimeout), + "no responders": fmt.Errorf("build 1: %w", nats.ErrNoResponders), + "a closed bus": nats.ErrConnectionClosed, + "the lease lost": fmt.Errorf("%w: another controller holds it", chain.ErrNotHeld), + "a key moved": chain.ErrMoved, + "a deadline": context.DeadlineExceeded, + } { + if !Unreachable(err) { + t.Errorf("%s is read as an answer", name) + } + } + for name, err := range map[string]error{ + "a value too large": chain.ErrTooLarge, + "a key not taken": chain.ErrBadKey, + "a plan moved": ErrPlanMoved, + "a key that exists": chain.ErrExists, + } { + if Unreachable(err) { + t.Errorf("%s is read as a bus being away: asking again gets the same answer", name) + } + } +} diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go index bd7ee1fd..1f37dfd5 100644 --- a/internal/inventory/chain_move.go +++ b/internal/inventory/chain_move.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "time" "github.com/jackc/pgx/v5" @@ -121,9 +122,97 @@ func (i *Inventory) movers() []mover { return n, err }, lock: `lock table build, build_request in share mode`, + }, { + kind: KindRollouts, + rows: func(ctx context.Context) ([]moveRow, int, error) { + plans, err := i.plans(ctx, `order by created`) + if err != nil { + return nil, 0, err + } + merges, err := i.merges(ctx, ``) + if err != nil { + return nil, 0, err + } + tablePlans := len(plans) + if plans, err = holdMerges(plans, merges); err != nil { + return nil, 0, err + } + out := make([]moveRow, 0, len(plans)) + for _, p := range plans { + body, err := json.Marshal(p) + if err != nil { + return nil, 0, err + } + out = append(out, moveRow{key: p.ID, value: p, size: len(body)}) + } + return out, tablePlans + len(merges), nil + }, + create: func(ctx context.Context, c *chain.Chain, row moveRow) error { + _, err := chain.RolloutsOf[Plan](c).Create(ctx, row.key, row.value.(Plan)) + return err + }, + keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { + return chain.RolloutsOf[Plan](c).Keys(ctx) + }, + count: func(ctx context.Context, tx pgx.Tx) (int, error) { + var n int + err := tx.QueryRow(ctx, `select (select count(*) from release_plan) + (select count(*) from batched_merge)`).Scan(&n) + return n, err + }, + lock: `lock table release_plan, batched_merge in share mode`, }} } +// holdMerges puts each batched merge into the record of the plan that answers it. A merge no plan answers — the plan_id +// of a merge a failed walk carried is null, or names a plan the store never kept — is held, flagged waiting, by the +// newest plan that lists it among its merges, else the record kept for the merges no plan answers: the record that held it +// last. +func holdMerges(plans []Plan, merges []BatchedMerge) ([]Plan, error) { + waitingAt := -1 + at := make(map[string]int, len(plans)) + for k, p := range plans { + at[p.ID] = k + } + newest := func(match func(Plan) bool) int { + best := -1 + for k, p := range plans { + if match(p) && (best < 0 || p.Created.After(plans[best].Created)) { + best = k + } + } + return best + } + for _, m := range merges { + held := heldOf(m, m.Heard) + holder, answered := at[m.Plan] + if m.Plan == "" || !answered { + held.Waiting = true + holder = newest(func(p Plan) bool { + if p.Delivery == nil { + return false + } + for _, listed := range p.Delivery.Merges { + if strings.EqualFold(listed.Repository, m.Repository) && listed.Commit == m.Commit { + return true + } + } + return false + }) + if holder < 0 { + // Held by the record kept for the merges no plan answers, made with the others. + if waitingAt < 0 { + plans = append(plans, Plan{ID: WaitingRecord, State: "merges", Created: time.Unix(0, 0).UTC(), Revision: 1, + Tiers: [][]string{}, Modules: map[string]*PlanModule{}, Note: "the merges no plan answers"}) + waitingAt = len(plans) - 1 + } + holder = waitingAt + } + } + plans[holder].Held = append(plans[holder].Held, held) + } + return plans, nil +} + // MoveChain moves every kind not moved yet, in order, and says what it did. It needs the chain attached and the lease // held (the chain's writes ask). It stops at the first kind that fails. func (i *Inventory) MoveChain(ctx context.Context, say func(format string, args ...any)) ([]MoveReport, error) { diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go index 6cd193cb..febda177 100644 --- a/internal/inventory/chain_move_test.go +++ b/internal/inventory/chain_move_test.go @@ -422,3 +422,104 @@ func TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain(t *testin t.Fatalf("the request lost its words: %+v", done) } } + +// **The move carries every plan with the merges it answers, and the plans read the same from the bucket as from the +// table** (ADR 0408 §8, ADR 0276, issue 1092): a batch with a merge, a walk that was built and rolled, a merge waiting +// for a walk of its own, and a merge whose plan the store never kept — each in the record of the plan that held it last. +func TestTheMoveCarriesThePlansAndTheirMergesAndTheyReadTheSame(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + batch := aBatch("plan-1", time.Now().Add(-3*time.Hour).UTC()) + walk := aBatch("plan-2", time.Now().Add(-2*time.Hour).UTC()) + walk.State, walk.Tier, walk.Tiers = PlanRolling, 1, [][]string{{"a"}, {"b"}} + walk.Modules = map[string]*PlanModule{"a": {State: "built"}, "b": {}} + walk.Delivery = &PlanDelivery{Awaits: "mesh-delivery", Merges: []PlanMerge{{Repository: "novox/mesh-catalog", Commit: "m-waiting"}}} + for _, p := range []*Plan{batch, walk} { + if err := inv.SavePlan(ctx, p); err != nil { + t.Fatal(err) + } + } + walk.Note = "second save" + if err := inv.SavePlan(ctx, walk); err != nil { + t.Fatal(err) + } + for _, m := range []BatchedMerge{aMerge("m-batch", "plan-1"), aMerge("m-walk", "plan-2"), aMerge("m-dangling", "plan-never-kept")} { + if added, err := inv.AddMerge(ctx, m); err != nil || !added { + t.Fatalf("added %v: %v", added, err) + } + } + waiting := aMerge("m-waiting", "plan-2") + if _, err := inv.AddMerge(ctx, waiting); err != nil { + t.Fatal(err) + } + if err := inv.AnswerMerge(ctx, waiting.Repository, waiting.Commit, "", true); err != nil { + t.Fatal(err) + } + wasPlans, err := inv.RecentPlans(ctx, 10) + if err != nil || len(wasPlans) != 2 { + t.Fatalf("the table holds %d plans: %v", len(wasPlans), err) + } + wasMerge, _, _ := inv.MergeOf(ctx, "novox/mesh-catalog", "m-walk") + + reports, err := inv.MoveChain(ctx, nil) + if err != nil { + t.Fatal(err) + } + var moved MoveReport + for _, r := range reports { + if r.Kind == KindRollouts { + moved = r + } + } + // Two plans, and the record of the merge whose plan the store never kept. + if moved.Rows != 3 || moved.TableRows != 6 || moved.Written != 3 { + t.Fatalf("the move of the rollouts reported %+v", moved) + } + nowPlans, err := inv.RecentPlans(ctx, 10) + if err != nil || len(nowPlans) != 2 { + t.Fatalf("the bucket holds %d plans: %v", len(nowPlans), err) + } + for k := range wasPlans { + was, now := wasPlans[k], nowPlans[k] + if was.ID != now.ID || was.State != now.State || was.Tier != now.Tier || was.Revision != now.Revision || was.Note != now.Note || + !was.Created.Equal(now.Created) || !reflect.DeepEqual(was.Tiers, now.Tiers) || len(was.Modules) != len(now.Modules) { + t.Errorf("plan %s was %+v and is %+v", was.ID, was, now) + } + } + if open, _ := inv.OpenPlans(ctx); len(open) != 1 || open[0].ID != "plan-2" { + t.Fatalf("open plans %+v", open) + } + if batches, _ := inv.Batches(ctx); len(batches) != 1 || batches[0].ID != "plan-1" { + t.Fatalf("batches %+v", batches) + } + if got, _ := inv.MergesOf(ctx, "plan-1"); len(got) != 1 || got[0].Commit != "m-batch" { + t.Fatalf("the batch answers %+v", got) + } + nowMerge, known, _ := inv.MergeOf(ctx, "novox/mesh-catalog", "m-walk") + if !known || nowMerge.Plan != "plan-2" || nowMerge.Branch != wasMerge.Branch || !nowMerge.Merged.Equal(wasMerge.Merged) || + !nowMerge.Heard.Equal(wasMerge.Heard) || string(nowMerge.Event) != string(wasMerge.Event) { + t.Fatalf("a merge was %+v and is %+v", wasMerge, nowMerge) + } + alone, _ := inv.AloneMerges(ctx) + if len(alone) != 1 || alone[0].Commit != "m-waiting" || alone[0].Plan != "" { + t.Fatalf("waiting: %+v", alone) + } + if dangling, known, _ := inv.MergeOf(ctx, "novox/mesh-catalog", "m-dangling"); !known || dangling.Plan != "" || dangling.Alone { + t.Fatalf("a merge whose plan was never kept reads as %+v %v", dangling, known) + } + if n := rowsInTheTable(t, inv, "release_plan"); n != "2" { + t.Fatalf("the table is %s plans", n) + } + // From now on a plan is saved to the bucket and the table takes none. + later := aBatch("plan-3", time.Now().UTC()) + if err := inv.SavePlan(ctx, later); err != nil { + t.Fatal(err) + } + if n := rowsInTheTable(t, inv, "release_plan"); n != "2" { + t.Fatalf("the table took a plan after the move: %s", n) + } + if again, err := inv.MoveChain(ctx, nil); err != nil || len(again) != 2 || !again[0].Already || !again[1].Already { + t.Fatalf("the second move %+v: %v", again, err) + } +} diff --git a/internal/inventory/chain_rollouts.go b/internal/inventory/chain_rollouts.go new file mode 100644 index 00000000..98a49f86 --- /dev/null +++ b/internal/inventory/chain_rollouts.go @@ -0,0 +1,939 @@ +package inventory + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// Rollouts on the bus (novox/hq ADR 0408, ADR 0276, issue 1092): once the kind `rollouts` is moved (chain.go) every plan +// and batch is a key of the rollouts bucket under its id, written by compare-and-set, and the merges a batch or walk +// answers are held in its record. The tables are read when the kind is not moved, by the methods these stand beside. +// +// **Two counters.** The bucket's revision is the compare-and-set of every write. The plan's own Revision, kept in its value, +// counts the saves of the plan: a save refuses (ErrPlanMoved) a plan another save moved since it was read, and takes no +// notice of a merge moving in or out of the record, which a copy read before could not know of. Held is written by the +// merge operations alone; a save keeps what the record holds. + +// rolloutView is the rollouts bucket in memory, kept current by one watch (issue 1092): the survey of the plans, the +// batches and the merges is read from here, never from the bucket one key at a time. A write is still made by +// compare-and-set on the bucket, so a stale view costs a refused write and a read again, never a wrong one. +type rolloutView struct { + chain *chain.Chain + mu sync.RWMutex + entries map[string]viewEntry + ready chan struct{} + once sync.Once + err error + dead atomic.Bool + cancel context.CancelFunc +} + +// viewEntry is a record as the bucket holds it, with what the surveys ask of it read out. +type viewEntry struct { + raw []byte + rev uint64 + sum planSummary +} + +// planSummary is the part of a plan the surveys filter and sort by. +type planSummary struct { + ID string `json:"id"` + State string `json:"state"` + Created time.Time `json:"created"` + Repository string `json:"repository"` + Branch string `json:"branch"` + Merged time.Time `json:"merged"` + Release json.RawMessage `json:"release"` + Held []chain.HeldMerge `json:"held"` +} + +func (s planSummary) open() bool { return s.State == PlanBuilding || s.State == PlanRolling } + +func (s planSummary) isRelease() bool { + r := strings.TrimSpace(string(s.Release)) + return r != "" && r != "null" +} + +func newRolloutView(c *chain.Chain) *rolloutView { + return &rolloutView{chain: c, entries: map[string]viewEntry{}, ready: make(chan struct{})} +} + +func (v *rolloutView) start() { + ctx, cancel := context.WithCancel(context.Background()) + v.cancel = cancel + go func() { + w, err := chain.RolloutsOf[Plan](v.chain).Watch(ctx) + if err != nil { + v.fail(fmt.Errorf("the rollouts cannot be watched: %w", err)) + return + } + for { + select { + case <-ctx.Done(): + v.fail(ctx.Err()) + return + case e, open := <-w: + if !open { + v.fail(errors.New("the rollouts bucket stopped answering")) + return + } + if e.Caught { + v.once.Do(func() { close(v.ready) }) + continue + } + v.apply(e.Key, e.Value, e.Revision) + } + } + }() +} + +// fail ends the view: it answers an error to whoever waits for it to fill, and is made again by the next read. +func (v *rolloutView) fail(err error) { + v.mu.Lock() + if v.err == nil { + v.err = err + } + v.mu.Unlock() + v.dead.Store(true) + v.once.Do(func() { close(v.ready) }) +} + +func (v *rolloutView) stop() { + if v.cancel != nil { + v.cancel() + } +} + +// apply keeps a record the watch handed over or a write of this process made, unless the view holds a newer one. +func (v *rolloutView) apply(key string, raw []byte, rev uint64) { + var sum planSummary + if err := json.Unmarshal(raw, &sum); err != nil || sum.ID == "" { + sum = planSummary{ID: key} + } + v.mu.Lock() + defer v.mu.Unlock() + if old, has := v.entries[key]; has && old.rev >= rev { + return + } + v.entries[key] = viewEntry{raw: append([]byte(nil), raw...), rev: rev, sum: sum} +} + +func (v *rolloutView) list() []viewEntry { + v.mu.RLock() + defer v.mu.RUnlock() + out := make([]viewEntry, 0, len(v.entries)) + for _, e := range v.entries { + out = append(out, e) + } + return out +} + +// rolloutView is the view of the chain's rollouts, started at the first read and ready when it holds the bucket whole. +func (i *Inventory) rolloutView(ctx context.Context, c *chain.Chain) (*rolloutView, error) { + i.moved.mu.Lock() + v := i.moved.view + if v == nil || v.chain != c || v.dead.Load() { + if v != nil { + v.stop() + } + v = newRolloutView(c) + i.moved.view = v + v.start() + } + i.moved.mu.Unlock() + select { + case <-v.ready: + case <-ctx.Done(): + return nil, ctx.Err() + } + v.mu.RLock() + defer v.mu.RUnlock() + if v.err != nil && len(v.entries) == 0 { + return nil, fmt.Errorf("the rollouts cannot be read from the bus: %w", v.err) + } + return v, nil +} + +// stopViews ends the view's watch: the inventory is closing. +func (i *Inventory) stopViews() { + i.moved.mu.Lock() + defer i.moved.mu.Unlock() + if i.moved.view != nil { + i.moved.view.stop() + i.moved.view = nil + } +} + +func normalizePlan(p *Plan) { + if p.Modules == nil { + p.Modules = map[string]*PlanModule{} + } + if p.TierEntered.IsZero() { + p.TierEntered = p.Created + } + if !p.Merged.IsZero() { + p.Merged = p.Merged.UTC() + } +} + +func planOf(e viewEntry) (Plan, error) { + var p Plan + if err := json.Unmarshal(e.raw, &p); err != nil { + return Plan{}, fmt.Errorf("the plan %s on the bus cannot be read: %w", e.sum.ID, err) + } + normalizePlan(&p) + return p, nil +} + +func sortCreated(entries []viewEntry) { + sort.SliceStable(entries, func(a, b int) bool { + if !entries[a].sum.Created.Equal(entries[b].sum.Created) { + return entries[a].sum.Created.Before(entries[b].sum.Created) + } + return entries[a].sum.ID < entries[b].sum.ID + }) +} + +func sortCreatedNewest(entries []viewEntry) { + sort.SliceStable(entries, func(a, b int) bool { + if !entries[a].sum.Created.Equal(entries[b].sum.Created) { + return entries[a].sum.Created.After(entries[b].sum.Created) + } + return entries[a].sum.ID > entries[b].sum.ID + }) +} + +// chainPlans is the plans that pass keep, in an order, read from the view. +func (i *Inventory) chainPlans(ctx context.Context, c *chain.Chain, keep func(viewEntry) bool, order func([]viewEntry)) ([]Plan, error) { + v, err := i.rolloutView(ctx, c) + if err != nil { + return nil, err + } + var chosen []viewEntry + for _, e := range v.list() { + if e.sum.ID != WaitingRecord && keep(e) { + chosen = append(chosen, e) + } + } + order(chosen) + out := make([]Plan, 0, len(chosen)) + for _, e := range chosen { + p, err := planOf(e) + if err != nil { + return nil, err + } + out = append(out, p) + } + return out, nil +} + +func (i *Inventory) chainNewestMergeOf(ctx context.Context, c *chain.Chain, repository, branch string) (Plan, bool, error) { + v, err := i.rolloutView(ctx, c) + if err != nil { + return Plan{}, false, err + } + var chosen []viewEntry + for _, e := range v.list() { + if strings.EqualFold(e.sum.Repository, repository) && e.sum.Branch == branch && !e.sum.Merged.IsZero() && !e.sum.isRelease() { + chosen = append(chosen, e) + } + } + sort.SliceStable(chosen, func(a, b int) bool { + if !chosen[a].sum.Merged.Equal(chosen[b].sum.Merged) { + return chosen[a].sum.Merged.After(chosen[b].sum.Merged) + } + return chosen[a].sum.Created.After(chosen[b].sum.Created) + }) + if len(chosen) == 0 { + return Plan{}, false, nil + } + p, err := planOf(chosen[0]) + return p, err == nil, err +} + +// chainPlanByID is one plan, read from the bucket itself: the caller of the verb wants the plan as it is, and the view +// learns of it. +func (i *Inventory) chainPlanByID(ctx context.Context, c *chain.Chain, id string) (Plan, error) { + if !chain.ValidKey(id) { + // No plan has an id the bus takes no spelling of; it is a plain id no plan has. + return Plan{}, fmt.Errorf("no plan %s", id) + } + e, found, err := chain.RolloutsOf[Plan](c).GetRaw(ctx, id) + if err != nil { + return Plan{}, err + } + if !found { + return Plan{}, fmt.Errorf("no plan %s", id) + } + if v, err := i.rolloutView(ctx, c); err == nil { + v.apply(id, e.Value, e.Revision) + } + return planOf(viewEntry{raw: e.Value, rev: e.Revision, sum: planSummary{ID: id}}) +} + +// ---- the merges + +// mergeHolder is a record's copy of a merge. +type mergeHolder struct { + plan string + m chain.HeldMerge + rev uint64 +} + +func mergeKey(repository, commit string) string { return strings.ToLower(repository) + "|" + commit } + +// mergeHolders is every record's copy of each merge, by the merge's key. +func mergeHolders(entries []viewEntry) map[string][]mergeHolder { + out := map[string][]mergeHolder{} + for _, e := range entries { + for _, m := range e.sum.Held { + k := mergeKey(m.Repository, m.Commit) + out[k] = append(out[k], mergeHolder{plan: e.sum.ID, m: m, rev: e.rev}) + } + } + return out +} + +// winner is the copy of a merge that counts: the one that entered its record last (ADR 0408, issue 1092 — a merge in two +// records is the half-done move of one, and the record it entered last is the one it moved to), a merge a plan answers +// before one no plan does, then the record written last. +func winner(copies []mergeHolder) mergeHolder { + best := copies[0] + for _, c := range copies[1:] { + switch { + case c.m.Placed.After(best.m.Placed): + best = c + case c.m.Placed.Equal(best.m.Placed) && best.m.Waiting && !c.m.Waiting: + best = c + case c.m.Placed.Equal(best.m.Placed) && best.m.Waiting == c.m.Waiting && c.rev > best.rev: + best = c + } + } + return best +} + +func winners(entries []viewEntry) map[string]mergeHolder { + out := map[string]mergeHolder{} + for k, copies := range mergeHolders(entries) { + out[k] = winner(copies) + } + return out +} + +func batchedOf(h mergeHolder) BatchedMerge { + b := BatchedMerge{Repository: strings.ToLower(h.m.Repository), Branch: h.m.Branch, Commit: h.m.Commit, + Heard: h.m.Heard.UTC(), Event: h.m.Event, Plan: h.plan, Alone: h.m.Alone} + if !h.m.Merged.IsZero() { + b.Merged = h.m.Merged.UTC() + } + if h.m.Waiting { + b.Plan = "" + } + return b +} + +func heldOf(m BatchedMerge, now time.Time) chain.HeldMerge { + h := chain.HeldMerge{Repository: strings.ToLower(m.Repository), Branch: m.Branch, Commit: m.Commit, Heard: m.Heard.UTC(), + Event: m.Event, Alone: m.Alone, Placed: now.UTC()} + if !m.Merged.IsZero() { + h.Merged = m.Merged.UTC() + } + return h +} + +func (i *Inventory) chainWinners(ctx context.Context, c *chain.Chain) (map[string]mergeHolder, error) { + v, err := i.rolloutView(ctx, c) + if err != nil { + return nil, err + } + return winners(v.list()), nil +} + +func (i *Inventory) chainMergeOf(ctx context.Context, c *chain.Chain, repository, commit string) (BatchedMerge, bool, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return BatchedMerge{}, false, err + } + h, ok := w[mergeKey(repository, commit)] + if !ok { + return BatchedMerge{}, false, nil + } + return batchedOf(h), true, nil +} + +// merged orders merges by the time they were made, those whose time is not known last, then by when they were heard. +func sortMerges(ms []BatchedMerge, newest bool) { + sort.SliceStable(ms, func(a, b int) bool { + x, y := ms[a], ms[b] + switch { + case x.Merged.IsZero() != y.Merged.IsZero(): + return !x.Merged.IsZero() + case !x.Merged.Equal(y.Merged): + if newest { + return x.Merged.After(y.Merged) + } + return x.Merged.Before(y.Merged) + } + if newest { + return x.Heard.After(y.Heard) + } + return x.Heard.Before(y.Heard) + }) +} + +func (i *Inventory) chainMergesOf(ctx context.Context, c *chain.Chain, plan string) ([]BatchedMerge, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return nil, err + } + var out []BatchedMerge + for _, h := range w { + if h.plan == plan && !h.m.Waiting { + out = append(out, batchedOf(h)) + } + } + sortMerges(out, false) + return out, nil +} + +func (i *Inventory) chainLaterMergesOf(ctx context.Context, c *chain.Chain, repository, branch string, after time.Time) ([]BatchedMerge, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return nil, err + } + var out []BatchedMerge + for _, h := range w { + if !h.m.Waiting && strings.EqualFold(h.m.Repository, repository) && h.m.Branch == branch && h.m.Merged.After(after) { + out = append(out, batchedOf(h)) + } + } + sort.SliceStable(out, func(a, b int) bool { return out[a].Merged.After(out[b].Merged) }) + return out, nil +} + +func (i *Inventory) chainKeptOnBranch(ctx context.Context, c *chain.Chain, repository, branch string) (bool, time.Time, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return false, time.Time{}, err + } + var first time.Time + for _, h := range w { + if strings.EqualFold(h.m.Repository, repository) && h.m.Branch == branch && (first.IsZero() || h.m.Heard.Before(first)) { + first = h.m.Heard + } + } + if first.IsZero() { + return false, time.Time{}, nil + } + return true, first.UTC(), nil +} + +func (i *Inventory) chainAloneMerges(ctx context.Context, c *chain.Chain) ([]BatchedMerge, error) { + w, err := i.chainWinners(ctx, c) + if err != nil { + return nil, err + } + var out []BatchedMerge + for _, h := range w { + if h.m.Waiting && h.m.Alone { + out = append(out, batchedOf(h)) + } + } + sortMerges(out, true) + return out, nil +} + +// pendingMerge is a merge added to a plan that is not saved yet: carried until that plan's save writes it. +type pendingMerge struct { + plan string + m chain.HeldMerge +} + +// chainAddMerge keeps a merge for a plan. **When the plan has a record the merge is written into it at once**; when it has +// none — a new batch, saved by the caller next — the merge is carried and written by that plan's save. The forge's event is +// acknowledged only after the caller's write succeeded, and handed over again if it did not (ADR 0407 §3). +func (i *Inventory) chainAddMerge(ctx context.Context, c *chain.Chain, m BatchedMerge) (bool, error) { + waiting := m.Plan == "" + if waiting { + // A merge no plan answers is held, flagged, by the record kept for those. + m.Plan = WaitingRecord + if err := i.ensureWaitingRecord(ctx, c); err != nil { + return false, err + } + } + v, err := i.rolloutView(ctx, c) + if err != nil { + return false, err + } + key := mergeKey(m.Repository, m.Commit) + if _, known := winners(v.list())[key]; known { + return false, nil + } + held := heldOf(m, time.Now()) + held.Waiting = waiting + found, err := i.editRecord(ctx, c, m.Plan, func(h []chain.HeldMerge) ([]chain.HeldMerge, bool) { + for _, x := range h { + if mergeKey(x.Repository, x.Commit) == key { + return h, false + } + } + return append(h, held), true + }) + if err != nil { + return false, err + } + if !found { + i.moved.mu.Lock() + if i.moved.pending == nil { + i.moved.pending = map[string]pendingMerge{} + } + i.moved.pending[key] = pendingMerge{plan: m.Plan, m: held} + i.moved.mu.Unlock() + } + return true, nil +} + +// WaitingRecord is the record that holds the merges no plan has ever answered: a merge kept with no plan, which a store +// kept as a row with a null plan. It is no plan: the surveys of plans leave it out. +const WaitingRecord = "waiting-merges" + +// ensureWaitingRecord makes the record of the merges no plan answers, once. +func (i *Inventory) ensureWaitingRecord(ctx context.Context, c *chain.Chain) error { + b := chain.RolloutsOf[Plan](c) + if _, found, err := b.GetRaw(ctx, WaitingRecord); err != nil || found { + return err + } + body, err := json.Marshal(Plan{ID: WaitingRecord, State: "merges", Created: time.Unix(0, 0).UTC(), Revision: 1, + Tiers: [][]string{}, Modules: map[string]*PlanModule{}, Note: "the merges no plan answers"}) + if err != nil { + return err + } + rev, err := b.CreateRaw(ctx, WaitingRecord, body) + if errors.Is(err, chain.ErrExists) { + return nil + } + if err != nil { + return err + } + if v, verr := i.rolloutView(ctx, c); verr == nil { + v.apply(WaitingRecord, body, rev) + } + return nil +} + +// ErrMergeUnsaved is a merge moved to a plan that has no record to hold it. +var ErrMergeUnsaved = errors.New("a merge was moved to a plan that has not been saved") + +// editRecord changes the merges a plan's record holds and nothing else of it, by compare-and-set on the bucket, read again +// when it moved. It answers false when the plan has no record. +func (i *Inventory) editRecord(ctx context.Context, c *chain.Chain, id string, edit func([]chain.HeldMerge) ([]chain.HeldMerge, bool)) (bool, error) { + b := chain.RolloutsOf[Plan](c) + for try := 0; try < casTries; try++ { + e, found, err := b.GetRaw(ctx, id) + if err != nil || !found { + return false, err + } + var record map[string]json.RawMessage + if err := json.Unmarshal(e.Value, &record); err != nil { + return true, fmt.Errorf("the plan %s on the bus cannot be read: %w", id, err) + } + var held []chain.HeldMerge + if raw, has := record["held"]; has { + if err := json.Unmarshal(raw, &held); err != nil { + return true, fmt.Errorf("the merges of plan %s on the bus cannot be read: %w", id, err) + } + } + next, changed := edit(held) + if !changed { + return true, nil + } + if len(next) == 0 { + delete(record, "held") + } else { + raw, err := json.Marshal(next) + if err != nil { + return true, err + } + record["held"] = raw + } + body, err := json.Marshal(record) + if err != nil { + return true, err + } + rev, err := b.UpdateRaw(ctx, id, body, e.Revision) + if errors.Is(err, chain.ErrMoved) { + continue + } + if err != nil { + return true, fmt.Errorf("the merges of plan %s are not written: %w", id, err) + } + if v, verr := i.rolloutView(ctx, c); verr == nil { + v.apply(id, body, rev) + } + return true, nil + } + return true, fmt.Errorf("the merges of plan %s kept moving under the write", id) +} + +// recordEdit is what the merges of one record gain, lose and have flagged in one act. +type recordEdit struct { + add []chain.HeldMerge + drop map[string]bool + flag map[string]chain.HeldMerge +} + +func (e *recordEdit) apply(held []chain.HeldMerge) []chain.HeldMerge { + out := make([]chain.HeldMerge, 0, len(held)+len(e.add)) + for _, h := range held { + k := mergeKey(h.Repository, h.Commit) + if e.drop[k] { + continue + } + if f, has := e.flag[k]; has { + h.Waiting, h.Alone = f.Waiting, f.Alone + } + out = append(out, h) + } + for _, a := range e.add { + k := mergeKey(a.Repository, a.Commit) + dup := false + for _, h := range out { + if mergeKey(h.Repository, h.Commit) == k { + dup = true + break + } + } + if !dup { + out = append(out, a) + } + } + return out +} + +func (e *recordEdit) empty() bool { return len(e.add) == 0 && len(e.drop) == 0 && len(e.flag) == 0 } + +// ---- the save + +// tierLeftOf is the measurement a plan's save makes when it leaves a tier (planTierLeft, over the record the save +// replaces): when the plan entered the tier it is now at, and the duration of the tier it left, if it left one. +func tierLeftOf(old *Plan, p Plan, now time.Time) (time.Time, *Duration) { + if old == nil { + return now, nil // a new plan enters its first tier now + } + if old.State == PlanAssembling || old.State == PlanQueued { + return now, nil // a batch cut into a walk enters its first tier now (novox/hq ADR 0276) + } + since := old.TierEntered + if since.IsZero() { + since = old.Created + } + wasOpen := old.State == PlanBuilding || old.State == PlanRolling + if !wasOpen || (old.Tier == p.Tier && p.Open()) { + return since, nil // still in the tier, or already ended + } + var modules []string + if old.Tier >= 0 && old.Tier < len(p.Tiers) { + modules = p.Tiers[old.Tier] + } + detail := fmt.Sprintf("plan %s, tier %d of %d (%v), left %s", p.ID, old.Tier, len(p.Tiers), modules, p.State) + if p.Open() { + detail = fmt.Sprintf("plan %s, tier %d of %d (%v), moved on to tier %d", p.ID, old.Tier, len(p.Tiers), modules, p.Tier) + } + return now, &Duration{Kind: DurationPlanTier, Subject: p.Repository, Ref: fmt.Sprintf("%s/tier-%d", p.ID, old.Tier), + Started: since, Took: now.Sub(since), Detail: detail} +} + +// chainSavePlans writes plans, as SavePlan does each, and gives merges to the plans that answer them. **Each plan by +// compare-and-set on its own Revision**, a copy read before another save refused with ErrPlanMoved before anything is +// written; **a merge moved to a plan is written into the receiving record first, then taken from the giving one**, so a +// crash between leaves it in both and never in neither, and the one it entered last counts (ADR 0408, issue 1092). +func (i *Inventory) chainSavePlans(ctx context.Context, c *chain.Chain, plans []*Plan, moves []MergeMove) (err error) { + epoch, err := i.actingEpoch(ctx) + if err != nil { + if len(plans) > 0 { + return fmt.Errorf("the plan for %s %s is not written: %w", plans[0].Repository, plans[0].Commit, err) + } + return err + } + v, err := i.rolloutView(ctx, c) + if err != nil { + return err + } + b := chain.RolloutsOf[Plan](c) + marked := time.Now() + undo := make([]func(), 0, len(plans)) + defer func() { + if err != nil { + for _, u := range undo { + u() + } + } + }() + for _, p := range plans { + undo = append(undo, p.markWalk(marked)) + } + + // Every plan still at the revision it was read at, before anything is written. + cur := make(map[string]chain.Kept[Plan], len(plans)) + for _, p := range plans { + k, found, err := b.Get(ctx, p.ID) + if err != nil { + return err + } + if found { + if k.Value.Revision != p.Revision { + return fmt.Errorf("the plan for %s %s (%s) is not written: %w", p.Repository, short(p.Commit), p.ID, ErrPlanMoved) + } + cur[p.ID] = k + } + } + + // What each record gains, loses and has flagged. + now := time.Now().UTC() + edits := map[string]*recordEdit{} + editOf := func(id string) *recordEdit { + if edits[id] == nil { + edits[id] = &recordEdit{drop: map[string]bool{}, flag: map[string]chain.HeldMerge{}} + } + return edits[id] + } + copies := mergeHolders(v.list()) + for _, m := range moves { + key := mergeKey(m.Repository, m.Commit) + held, has := copies[key] + if !has { + // A merge carried for a plan not saved yet goes where it is moved. + i.moved.mu.Lock() + if pm, pending := i.moved.pending[key]; pending && m.Plan != "" { + pm.plan, pm.m.Alone = m.Plan, m.Alone + i.moved.pending[key] = pm + } + i.moved.mu.Unlock() + continue + } + w := winner(held) + if m.Plan == "" { + f := w.m + f.Waiting, f.Alone = true, m.Alone + editOf(w.plan).flag[key] = f + continue + } + if w.plan == m.Plan { + f := w.m + f.Waiting, f.Alone = false, m.Alone + editOf(w.plan).flag[key] = f + continue + } + moved := w.m + moved.Waiting, moved.Alone, moved.Placed = false, m.Alone, now + editOf(m.Plan).add = append(editOf(m.Plan).add, moved) + for _, h := range held { + editOf(h.plan).drop[key] = true + } + } + + // The records not among the plans: receivers now, givers after. + for id, e := range edits { + if _, written := cur[id]; written || planNamed(plans, id) || len(e.add) == 0 { + continue + } + e := e + found, err := i.editRecord(ctx, c, id, func(h []chain.HeldMerge) ([]chain.HeldMerge, bool) { + return (&recordEdit{add: e.add}).apply(h), true + }) + if err != nil { + return err + } + if !found { + // Nothing is taken from the record that holds a merge to give it to a plan that has none. + return fmt.Errorf("%w: %s has no record to hold %d merge(s) moved to it", ErrMergeUnsaved, id, len(e.add)) + } + } + + // The plans, those that gain a merge first. + order := append([]*Plan(nil), plans...) + sort.SliceStable(order, func(a, b int) bool { + x, y := edits[order[a].ID], edits[order[b].ID] + return x != nil && len(x.add) > 0 && (y == nil || len(y.add) == 0) + }) + type written struct { + revision int64 + entered time.Time + held []chain.HeldMerge + duration *Duration + } + done := map[string]written{} + for _, p := range order { + var w written + for try := 0; ; try++ { + if try == casTries { + return fmt.Errorf("the plan for %s %s (%s) kept moving under the write", p.Repository, short(p.Commit), p.ID) + } + k, found := cur[p.ID] + if try > 0 { + fresh, still, gerr := b.Get(ctx, p.ID) + if gerr != nil { + return gerr + } + if still && fresh.Value.Revision != p.Revision { + return fmt.Errorf("the plan for %s %s (%s) is not written: %w", p.Repository, short(p.Commit), p.ID, ErrPlanMoved) + } + k, found = fresh, still + cur[p.ID] = k + } + record := *p + var old *Plan + var held []chain.HeldMerge + if found { + o := k.Value + old, held = &o, append([]chain.HeldMerge(nil), k.Value.Held...) + record.Revision = k.Value.Revision + 1 + } else { + record.Revision = 1 + } + if e := edits[p.ID]; e != nil { + held = e.apply(held) + } + i.moved.mu.Lock() + for key, pm := range i.moved.pending { + if pm.plan == p.ID { + held = (&recordEdit{add: []chain.HeldMerge{pm.m}}).apply(held) + _ = key + } + } + i.moved.mu.Unlock() + record.Held = held + entered, duration := tierLeftOf(old, *p, time.Now()) + record.TierEntered, record.Updated = entered, time.Now().UTC() + if epoch != nil { + record.Epoch = uint64(*epoch) + } else { + record.Epoch = 0 + } + body, merr := json.Marshal(record) + if merr != nil { + return merr + } + var rev uint64 + var werr error + if found { + rev, werr = b.UpdateRaw(ctx, p.ID, body, k.Revision) + } else { + rev, werr = createRaw(ctx, c, p.ID, body) + } + if errors.Is(werr, chain.ErrMoved) || (!found && errors.Is(werr, chain.ErrExists)) { + if !found { + return fmt.Errorf("the plan for %s %s (%s) is not written: %w", p.Repository, short(p.Commit), p.ID, ErrPlanMoved) + } + continue + } + if werr != nil { + return fmt.Errorf("the plan for %s %s (%s) is not written: %w", p.Repository, short(p.Commit), p.ID, werr) + } + v.apply(p.ID, body, rev) + w = written{revision: record.Revision, entered: entered, held: held, duration: duration} + break + } + done[p.ID] = w + } + + // The records that gave a merge up, now it is held by the one it went to. + for id, e := range edits { + if planNamed(plans, id) || (len(e.drop) == 0 && len(e.flag) == 0) { + continue + } + e := e + if _, err := i.editRecord(ctx, c, id, func(h []chain.HeldMerge) ([]chain.HeldMerge, bool) { + return (&recordEdit{drop: e.drop, flag: e.flag}).apply(h), true + }); err != nil { + return err + } + } + + i.moved.mu.Lock() + for key, pm := range i.moved.pending { + if _, saved := done[pm.plan]; saved { + delete(i.moved.pending, key) + } + } + i.moved.mu.Unlock() + for _, p := range plans { + w := done[p.ID] + p.Revision, p.TierEntered, p.Held = w.revision, w.entered, w.held + if epoch != nil { + p.Epoch = uint64(*epoch) + } else { + p.Epoch = 0 + } + if w.duration != nil { + // After the write: the key of the duration is the plan and its tier, so a measurement made twice is one. + record := i.RecordDuration + if i.recorded != nil { + record = i.recorded + } + if derr := record(ctx, *w.duration); derr != nil { + fmt.Printf("the time plan %s spent in a tier is not recorded: %v\n", p.ID, derr) + } + } + if PlanSaved != nil { + PlanSaved(i, *p) + } + } + return nil +} + +func planNamed(plans []*Plan, id string) bool { + for _, p := range plans { + if p.ID == id { + return true + } + } + return false +} + +// createRaw writes a plan's first record. +func createRaw(ctx context.Context, c *chain.Chain, id string, body []byte) (uint64, error) { + return chain.RolloutsOf[json.RawMessage](c).CreateRaw(ctx, id, body) +} + +// RepairMerges takes a merge found in two records out of the one it entered first: the half-done move of a crash, repaired +// at the next cut (ADR 0408, issue 1092). It says how many it repaired. +func (i *Inventory) RepairMerges(ctx context.Context) (int, error) { + c, on, err := i.ChainReads(ctx, KindRollouts) + if err != nil || !on { + return 0, err + } + v, err := i.rolloutView(ctx, c) + if err != nil { + return 0, err + } + repaired := 0 + for key, copies := range mergeHolders(v.list()) { + if len(copies) < 2 { + continue + } + w := winner(copies) + for _, h := range copies { + if h.plan == w.plan { + continue + } + key := key + if _, err := i.editRecord(ctx, c, h.plan, func(held []chain.HeldMerge) ([]chain.HeldMerge, bool) { + return (&recordEdit{drop: map[string]bool{key: true}}).apply(held), true + }); err != nil { + return repaired, err + } + repaired++ + } + } + return repaired, nil +} diff --git a/internal/inventory/chain_rollouts_test.go b/internal/inventory/chain_rollouts_test.go new file mode 100644 index 00000000..dffeeb17 --- /dev/null +++ b/internal/inventory/chain_rollouts_test.go @@ -0,0 +1,352 @@ +package inventory + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// anInventoryOnTheChain is an inventory with no store and a chain in memory: the chain's own paths, which never reach the +// store but for the durations a save measures, taken here by nothing. +func anInventoryOnTheChain(t *testing.T) (*Inventory, *chain.Chain) { + t.Helper() + c := chain.InMemory(nil) + inv := &Inventory{moved: &movedKinds{}, recorded: func(context.Context, Duration) error { return nil }} + inv.UseChain(c) + t.Cleanup(inv.stopViews) + return inv, c +} + +var plansAt = time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + +func aBatch(id string, at time.Time) *Plan { + return &Plan{ID: id, Repository: "novox/mesh-catalog", Branch: "main", Commit: "c-" + id, Created: at, State: PlanAssembling, + Tiers: [][]string{}, Modules: map[string]*PlanModule{}} +} + +func aMerge(commit, plan string) BatchedMerge { + return BatchedMerge{Repository: "Novox/Mesh-Catalog", Branch: "main", Commit: commit, Merged: plansAt, Heard: plansAt.Add(time.Second), + Event: []byte(`{"commit":"` + commit + `"}`), Plan: plan} +} + +// **A plan is kept under its id; a save of a copy another save has moved since is refused and writes nothing** (ADR 0408 +// §2, issue 1092): the plan's own Revision counts its saves, 1 for the first. +func TestAPlanIsKeptUnderItsIdAndASaveOfAStaleCopyIsRefused(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := context.Background() + p := aBatch("plan-1", plansAt) + if err := inv.chainSavePlans(ctx, c, []*Plan{p}, nil); err != nil { + t.Fatal(err) + } + if p.Revision != 1 { + t.Fatalf("a new plan is at revision %d", p.Revision) + } + stale := *p + p.Note = "moved on" + if err := inv.chainSavePlans(ctx, c, []*Plan{p}, nil); err != nil || p.Revision != 2 { + t.Fatalf("the second save: revision %d, %v", p.Revision, err) + } + stale.Note = "laid over" + if err := inv.chainSavePlans(ctx, c, []*Plan{&stale}, nil); !errors.Is(err, ErrPlanMoved) { + t.Fatalf("a stale copy was saved: %v", err) + } + got, err := inv.chainPlanByID(ctx, c, "plan-1") + if err != nil || got.Note != "moved on" || got.Revision != 2 { + t.Fatalf("the plan is %+v: %v", got, err) + } + // A plan new to the bucket that is already there is refused too. + again := aBatch("plan-1", plansAt) + if err := inv.chainSavePlans(ctx, c, []*Plan{again}, nil); !errors.Is(err, ErrPlanMoved) { + t.Fatalf("a new plan over a kept one: %v", err) + } + if _, err := inv.chainPlanByID(ctx, c, "plan-nobody"); err == nil || err.Error() != "no plan plan-nobody" { + t.Fatalf("a plan nobody kept: %v", err) + } +} + +// **A save keeps the merges the record holds, not the ones its copy read** (issue 1092): a merge that moved into the record +// after the copy was read is not laid back by the copy's save. +func TestASaveKeepsTheMergesTheRecordHoldsAndNotTheCopys(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := context.Background() + batch := aBatch("plan-1", plansAt) + if err := inv.chainSavePlans(ctx, c, []*Plan{batch}, nil); err != nil { + t.Fatal(err) + } + if added, err := inv.chainAddMerge(ctx, c, aMerge("m1", "plan-1")); err != nil || !added { + t.Fatalf("added %v: %v", added, err) + } + // The copy was read before the merge came in; saving it neither fails nor loses the merge. + batch.Note = "window closed" + if err := inv.chainSavePlans(ctx, c, []*Plan{batch}, nil); err != nil { + t.Fatalf("a copy read before a merge came in: %v", err) + } + merges, err := inv.chainMergesOf(ctx, c, "plan-1") + if err != nil || len(merges) != 1 || merges[0].Commit != "m1" || merges[0].Repository != "novox/mesh-catalog" { + t.Fatalf("the merges of the plan: %+v %v", merges, err) + } + if string(merges[0].Event) != `{"commit":"m1"}` || merges[0].Heard.IsZero() || merges[0].Plan != "plan-1" { + t.Fatalf("the merge came back as %+v", merges[0]) + } + // A merge heard is not heard twice, whatever the spelling of its repository. + if added, err := inv.chainAddMerge(ctx, c, aMerge("m1", "plan-1")); err != nil || added { + t.Fatalf("added again %v: %v", added, err) + } +} + +// **A merge added to a plan that is not saved is written by that plan's save, and by nothing else** (issue 1092): until +// then it is in no record — so the event that brought it is handed over again if the save fails. +func TestAMergeAddedToAPlanNotSavedYetIsWrittenByItsSave(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := context.Background() + if added, err := inv.chainAddMerge(ctx, c, aMerge("m1", "plan-new")); err != nil || !added { + t.Fatalf("added %v: %v", added, err) + } + if _, known, _ := inv.chainMergeOf(ctx, c, "novox/mesh-catalog", "m1"); known { + t.Fatal("a merge for a plan not saved is in a record") + } + // The save is refused: nothing is written, the merge is still carried. + held := false + away := chain.New(chain.NewMemory(), chain.NewMemory(), chain.NewMemory(), func(context.Context) error { + if held { + return nil + } + return errors.New("the lease is not held") + }) + other := &Inventory{moved: &movedKinds{}, recorded: inv.recorded} + other.UseChain(away) + t.Cleanup(other.stopViews) + other.moved.pending = inv.moved.pending + p := aBatch("plan-new", plansAt) + if err := other.chainSavePlans(ctx, away, []*Plan{p}, nil); !errors.Is(err, chain.ErrNotHeld) { + t.Fatalf("the save answered %v", err) + } + if _, known, _ := other.chainMergeOf(ctx, away, "novox/mesh-catalog", "m1"); known { + t.Fatal("a refused save left the merge in a record") + } + held = true + if err := other.chainSavePlans(ctx, away, []*Plan{p}, nil); err != nil { + t.Fatal(err) + } + if m, known, _ := other.chainMergeOf(ctx, away, "novox/mesh-catalog", "m1"); !known || m.Plan != "plan-new" { + t.Fatalf("the save did not write the merge: %+v %v", m, known) + } +} + +// failingOn is a bucket that refuses to write one key: the second half of a move, interrupted. +type failingOn struct { + chain.Backend + key string + fail *atomic.Bool +} + +func (f failingOn) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + if key == f.key && f.fail.Load() { + return 0, errors.New("the bus went away") + } + return f.Backend.Update(ctx, key, value, revision) +} + +// **A merge moved to a plan is written to the receiving record first, then taken from the giving one; one found in both is +// read from the one it entered last and repaired** (issue 1092, ADR 0299 §7). +func TestAMergeMovedToAPlanIsWrittenToItFirstAndRepairedIfTheMoveStopped(t *testing.T) { + c := chain.InMemory(nil) + var fail atomic.Bool + rollouts := failingOn{Backend: chain.NewMemory(), key: "plan-a", fail: &fail} + c = chain.New(chain.NewMemory(), rollouts, chain.NewMemory(), func(context.Context) error { return nil }) + inv := &Inventory{moved: &movedKinds{}, recorded: func(context.Context, Duration) error { return nil }} + inv.UseChain(c) + t.Cleanup(inv.stopViews) + ctx := context.Background() + a, b := aBatch("plan-a", plansAt), aBatch("plan-b", plansAt.Add(time.Minute)) + for _, p := range []*Plan{a, b} { + if err := inv.chainSavePlans(ctx, c, []*Plan{p}, nil); err != nil { + t.Fatal(err) + } + } + if _, err := inv.chainAddMerge(ctx, c, aMerge("m1", "plan-a")); err != nil { + t.Fatal(err) + } + + // The move stops before the giving record is written: the merge is in both. + fail.Store(true) + err := inv.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: "novox/mesh-catalog", Commit: "m1", Plan: "plan-b"}}) + if err == nil { + t.Fatal("the move was said to be made with the giving record unwritten") + } + fail.Store(false) + if m, _, _ := inv.chainMergeOf(ctx, c, "novox/mesh-catalog", "m1"); m.Plan != "plan-b" { + t.Fatalf("a merge in two records is read from %q, not the one it moved to", m.Plan) + } + if got, _ := inv.chainMergesOf(ctx, c, "plan-a"); len(got) != 0 { + t.Fatalf("the giving plan still answers %+v", got) + } + // The next cut repairs it. + v, _ := inv.rolloutView(ctx, c) + if n := len(mergeHolders(v.list())["novox/mesh-catalog|m1"]); n != 2 { + t.Fatalf("the merge is held by %d records, want the two of a stopped move", n) + } + inv.moved.mu.Lock() + inv.moved.kind = map[string]bool{KindRollouts: true} + inv.moved.mu.Unlock() + repaired := 0 + for k, copies := range mergeHolders(v.list()) { + if len(copies) < 2 { + continue + } + w := winner(copies) + for _, h := range copies { + if h.plan != w.plan { + key := k + if _, err := inv.editRecord(ctx, c, h.plan, func(held []chain.HeldMerge) ([]chain.HeldMerge, bool) { + return (&recordEdit{drop: map[string]bool{key: true}}).apply(held), true + }); err != nil { + t.Fatal(err) + } + repaired++ + } + } + } + if repaired != 1 { + t.Fatalf("repaired %d", repaired) + } + waitFor(t, func() bool { + v, _ := inv.rolloutView(ctx, c) + return len(mergeHolders(v.list())["novox/mesh-catalog|m1"]) == 1 + }) +} + +// waitFor polls a condition that a watch makes true, for a moment. +func waitFor(t *testing.T, ok func() bool) { + t.Helper() + for end := time.Now().Add(3 * time.Second); time.Now().Before(end); time.Sleep(5 * time.Millisecond) { + if ok() { + return + } + } + t.Fatal("the condition did not come true") +} + +// **A merge no plan answers stays, flagged, in the record that held it last, and is walked alone** (issue 1092): not in +// the merges of the plan, listed among the merges waiting, and answered again by the plan it is moved to. +func TestAWaitingMergeStaysInTheRecordThatHeldItLast(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := context.Background() + a, b := aBatch("plan-a", plansAt), aBatch("plan-b", plansAt.Add(time.Minute)) + for _, p := range []*Plan{a, b} { + if err := inv.chainSavePlans(ctx, c, []*Plan{p}, nil); err != nil { + t.Fatal(err) + } + } + if _, err := inv.chainAddMerge(ctx, c, aMerge("m1", "plan-a")); err != nil { + t.Fatal(err) + } + if err := inv.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: "novox/mesh-catalog", Commit: "m1", Plan: "", Alone: true}}); err != nil { + t.Fatal(err) + } + if got, _ := inv.chainMergesOf(ctx, c, "plan-a"); len(got) != 0 { + t.Fatalf("a waiting merge is answered by %+v", got) + } + alone, err := inv.chainAloneMerges(ctx, c) + if err != nil || len(alone) != 1 || alone[0].Commit != "m1" || alone[0].Plan != "" || !alone[0].Alone { + t.Fatalf("waiting: %+v %v", alone, err) + } + if m, known, _ := inv.chainMergeOf(ctx, c, "novox/mesh-catalog", "m1"); !known || m.Plan != "" { + t.Fatalf("a waiting merge reads as %+v %v", m, known) + } + if later, _ := inv.chainLaterMergesOf(ctx, c, "novox/mesh-catalog", "main", plansAt.Add(-time.Hour)); len(later) != 0 { + t.Fatalf("a waiting merge is among those a plan answers: %+v", later) + } + if kept, first, _ := inv.chainKeptOnBranch(ctx, c, "novox/mesh-catalog", "main"); !kept || first.IsZero() { + t.Fatalf("a waiting merge is not kept on its branch: %v %v", kept, first) + } + // Walked alone: the plan it goes to answers it, and the record that held it lets it go. + if err := inv.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: "novox/mesh-catalog", Commit: "m1", Plan: "plan-b", Alone: true}}); err != nil { + t.Fatal(err) + } + if got, _ := inv.chainMergesOf(ctx, c, "plan-b"); len(got) != 1 || !got[0].Alone { + t.Fatalf("plan-b answers %+v", got) + } + if left, _ := inv.chainAloneMerges(ctx, c); len(left) != 0 { + t.Fatalf("still waiting: %+v", left) + } + waitFor(t, func() bool { + v, _ := inv.rolloutView(ctx, c) + return len(mergeHolders(v.list())["novox/mesh-catalog|m1"]) == 1 + }) + // A move to a plan with no record is refused and takes nothing from the one that holds the merge. + err = inv.chainSavePlans(ctx, c, nil, []MergeMove{{Repository: "novox/mesh-catalog", Commit: "m1", Plan: "plan-nowhere"}}) + if !errors.Is(err, ErrMergeUnsaved) { + t.Fatalf("a move to a plan with no record: %v", err) + } + if got, _ := inv.chainMergesOf(ctx, c, "plan-b"); len(got) != 1 { + t.Fatalf("the merge was taken from plan-b: %+v", got) + } +} + +// countingBackend counts the reads a bucket answers. +type countingBackend struct { + chain.Backend + gets, keys atomic.Int32 +} + +func (c *countingBackend) Get(ctx context.Context, key string) (chain.Entry, bool, error) { + c.gets.Add(1) + return c.Backend.Get(ctx, key) +} + +func (c *countingBackend) Keys(ctx context.Context) ([]string, error) { + c.keys.Add(1) + return c.Backend.Keys(ctx) +} + +// **The surveys are read from a view kept by one watch, never from the bucket key by key** (issue 1092): the plans and the +// merges are asked for many times and the bucket is read for none of them; a write by another process reaches the view +// through its watch. +func TestThePlansAreReadFromAViewKeptByOneWatch(t *testing.T) { + counting := &countingBackend{Backend: chain.NewMemory()} + c := chain.New(chain.NewMemory(), counting, chain.NewMemory(), func(context.Context) error { return nil }) + inv := &Inventory{moved: &movedKinds{}, recorded: func(context.Context, Duration) error { return nil }} + inv.UseChain(c) + t.Cleanup(inv.stopViews) + ctx := context.Background() + for k, id := range []string{"plan-1", "plan-2", "plan-3"} { + p := aBatch(id, plansAt.Add(time.Duration(k)*time.Minute)) + if k > 0 { + p.State = PlanBuilding + } + if err := inv.chainSavePlans(ctx, c, []*Plan{p}, nil); err != nil { + t.Fatal(err) + } + } + before := counting.gets.Load() + for n := 0; n < 20; n++ { + open, err := inv.chainPlans(ctx, c, func(e viewEntry) bool { return e.sum.open() }, sortCreated) + if err != nil || len(open) != 2 || open[0].ID != "plan-2" { + t.Fatalf("open plans %+v: %v", open, err) + } + if _, err := inv.chainAloneMerges(ctx, c); err != nil { + t.Fatal(err) + } + } + if counting.gets.Load() != before || counting.keys.Load() != 0 { + t.Fatalf("the bucket was read %d times and its keys %d times for the surveys", counting.gets.Load()-before, counting.keys.Load()) + } + // Another process writes; the view follows. + other := &Inventory{moved: &movedKinds{}, recorded: inv.recorded} + other.UseChain(c) + t.Cleanup(other.stopViews) + late := aBatch("plan-4", plansAt.Add(time.Hour)) + late.State = PlanRolling + if err := other.chainSavePlans(ctx, c, []*Plan{late}, nil); err != nil { + t.Fatal(err) + } + waitFor(t, func() bool { + open, _ := inv.chainPlans(ctx, c, func(e viewEntry) bool { return e.sum.open() }, sortCreated) + return len(open) == 3 + }) +} diff --git a/internal/inventory/fortest.go b/internal/inventory/fortest.go index 4da06c6b..a063b1e5 100644 --- a/internal/inventory/fortest.go +++ b/internal/inventory/fortest.go @@ -67,9 +67,12 @@ func ForTest(t *testing.T) *Inventory { } // The chain is on the bus in a mesh; in a test it is in memory, with every kind moved, as a mesh raised after this // release has them (ADR 0408, issue 1089). The tests of the move itself use ForTestBefore. - inv.UseChain(chain.InMemory(nil)) + c := chain.InMemory(nil) + inv.UseChain(c) + testChains.Store(os.Getenv(store.Variable(Name)), c) + t.Cleanup(func() { testChains.Delete(os.Getenv(store.Variable(Name))) }) if _, err := inv.store.Pool().Exec(t.Context(), - `insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0}`); err != nil { + `insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0, "rollouts": 0}`); err != nil { t.Fatal(err) } return inv @@ -130,7 +133,10 @@ func ForTestBefore(t *testing.T, number int) (*Inventory, func()) { t.Fatal(err) } // The chain is in memory and nothing is moved: the tests of a move seed the tables and move them. - inv.UseChain(chain.InMemory(nil)) + c := chain.InMemory(nil) + inv.UseChain(c) + testChains.Store(os.Getenv(store.Variable(Name)), c) + t.Cleanup(func() { testChains.Delete(os.Getenv(store.Variable(Name))) }) return inv, func() { t.Helper() if _, err := inv.store.Migrate(t.Context(), all); err != nil { diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index e422b75a..55cdfc27 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "fmt" + "os" "regexp" "slices" "sort" @@ -38,6 +39,8 @@ type Inventory struct { // in the tables as before. chain *chain.Chain moved *movedKinds + // recorded, when set, takes the durations a save of the chain measures in place of the store: a test with no store. + recorded func(context.Context, Duration) error } // Open connects to the inventory store. @@ -46,10 +49,18 @@ func Open(ctx context.Context) (*Inventory, error) { if err != nil { return nil, err } - return &Inventory{store: s, moved: &movedKinds{}}, nil + inv := &Inventory{store: s, moved: &movedKinds{}} + // A test's store has the chain a test made for it in memory, shared by every inventory opened on that store. + if c, ok := testChains.Load(os.Getenv(store.Variable(Name))); ok { + inv.chain = c.(*chain.Chain) + } + return inv, nil } -func (i *Inventory) Close() { i.store.Close() } +func (i *Inventory) Close() { + i.stopViews() + i.store.Close() +} // Ready waits for the database to answer. func (i *Inventory) Ready(ctx context.Context, within time.Duration) error { diff --git a/internal/inventory/plan_id_test.go b/internal/inventory/plan_id_test.go index 88cd0021..431643b2 100644 --- a/internal/inventory/plan_id_test.go +++ b/internal/inventory/plan_id_test.go @@ -3,14 +3,17 @@ package inventory import ( + "errors" "strings" "testing" "time" + + "github.com/novox/mesh-controller/internal/chain" ) -// **A plan's id is a value, never SQL** (novox/hq issue 561): the id reaches PlanByID from a caller of the +// **A plan's id is a value, never a query** (novox/hq issue 561): the id reaches PlanByID from a caller of the // controller's plans verb, so an id holding a quote is read as a plain id — no error from the store, and no -// other plan answered for it. +// other plan answered for it. On the bus an id is a key, and one the bus takes no spelling of is refused to be kept. func TestAPlanIdHoldingAQuoteIsReadAsAPlainId(t *testing.T) { inv := ForTest(t) ctx := t.Context() @@ -32,14 +35,11 @@ func TestAPlanIdHoldingAQuoteIsReadAsAPlainId(t *testing.T) { t.Fatalf("PlanByID(%q) failed with %q, not as a plain id no plan has (%q)", id, err, want) } } - // A plan whose id holds a quote is found by it, and by nothing else. + // A plan whose id holds a quote cannot be a key of the bus: refused by name, never rewritten into another id. quoted := p quoted.ID, quoted.Revision = "plan-'561'", 0 - if err := inv.SavePlan(ctx, "ed); err != nil { - t.Fatal(err) - } - if got, err := inv.PlanByID(ctx, quoted.ID); err != nil || got.ID != quoted.ID { - t.Fatalf("the plan with a quote in its id was not read back: %v %+v", err, got.ID) + if err := inv.SavePlan(ctx, "ed); !errors.Is(err, chain.ErrBadKey) { + t.Fatalf("a plan with a quote in its id was kept, or refused for another reason: %v", err) } if recent, err := inv.RecentPlans(ctx, 1); err != nil || len(recent) != 1 || !strings.HasPrefix(recent[0].ID, "plan-") { t.Fatalf("RecentPlans with a parameter: %v %+v", err, recent) diff --git a/internal/inventory/plans.go b/internal/inventory/plans.go index da3dad4a..8ad7a0d1 100644 --- a/internal/inventory/plans.go +++ b/internal/inventory/plans.go @@ -10,6 +10,8 @@ import ( "time" "github.com/jackc/pgx/v5" + + "github.com/novox/mesh-controller/internal/chain" ) // A Plan is what a merge produces (novox/hq ADR 0162): the modules it changed and everything @@ -66,6 +68,15 @@ type Plan struct { // per machine and module, what was asked and how it went, and the machines with no backup holder. Nil for a // walk that moves no module with an item protected by a backup, and for a walk kept before it was. Backups *PlanBackups `json:"backups,omitempty"` + // Links are what the rollout names of the chain (novox/hq ADR 0408 §1): the builds it carried, the machine steps it has + // taken and the assignments it changed, each with the build before and after. Kept in the rollout's record beside + // the fields above; empty in the controller's store. + chain.Links + // Held are the merges this plan or batch holds (novox/hq ADR 0276, ADR 0408): what the store kept as batched merges + // with this plan's id, and the merges no plan answers that it held last. **Written by the merge operations only**: a + // save of the plan keeps what the record holds, never what this copy read, so a copy read before a merge moved cannot + // lay it back (issue 1092). + Held []chain.HeldMerge `json:"held,omitempty"` // WalkSaid is, on the save that started or ended the walk, which of the two it was (novox/hq ADR 0308 // decision 1): set by the save, never kept, read by whoever says the save (`plan-moved`). Empty on every // other save. @@ -670,6 +681,11 @@ type MergeMove struct { // somebody else since it was read refuses the whole act with ErrPlanMoved. Each plan is said (PlanSaved) once // the transaction is committed, in order. func (i *Inventory) SavePlansMoving(ctx context.Context, plans []*Plan, moves []MergeMove) (err error) { + if c, on, cerr := i.ChainReads(ctx, KindRollouts); cerr != nil { + return cerr + } else if on { + return i.chainSavePlans(ctx, c, plans, moves) + } epoch, err := i.actingEpoch(ctx) if err != nil { if len(plans) > 0 { @@ -818,16 +834,35 @@ func mergedAt(t time.Time) *time.Time { // OpenPlans is every plan still being worked, oldest first. func (i *Inventory) OpenPlans(ctx context.Context) ([]Plan, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + return i.chainPlans(ctx, c, func(e viewEntry) bool { return e.sum.open() }, sortCreated) + } return i.plans(ctx, `where state in ('building', 'rolling') order by created`) } // PlansSince is every plan made after a moment, and every plan still being worked, oldest first. func (i *Inventory) PlansSince(ctx context.Context, since time.Time) ([]Plan, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + return i.chainPlans(ctx, c, func(e viewEntry) bool { return e.sum.Created.After(since) || e.sum.open() }, sortCreated) + } return i.plans(ctx, `where created > $1 or state in ('building', 'rolling') order by created`, since) } // RecentPlans is the last few plans, newest first, open or not — what the overview shows. func (i *Inventory) RecentPlans(ctx context.Context, limit int) ([]Plan, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return nil, err + } else if on { + plans, err := i.chainPlans(ctx, c, func(viewEntry) bool { return true }, sortCreatedNewest) + if err == nil && limit >= 0 && len(plans) > limit { + plans = plans[:limit] + } + return plans, err + } return i.plans(ctx, `order by created desc limit $1`, limit) } @@ -835,6 +870,11 @@ func (i *Inventory) RecentPlans(ctx context.Context, limit int) ([]Plan, error) // planned: the branch's newest commit the mesh knows of (novox/hq issue 349). False when no plan of it // recorded when its merge was made. func (i *Inventory) NewestMergeOf(ctx context.Context, repository, branch string) (Plan, bool, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return Plan{}, false, err + } else if on { + return i.chainNewestMergeOf(ctx, c, repository, branch) + } plans, err := i.plans(ctx, `where lower(repository) = lower($1) and branch = $2 and merged_at is not null and release is null order by merged_at desc, created desc limit 1`, repository, branch) if err != nil || len(plans) == 0 { @@ -846,6 +886,11 @@ func (i *Inventory) NewestMergeOf(ctx context.Context, repository, branch string // PlanByID is one plan. The id is a parameter, never part of the query (novox/hq issue 561): it comes from a // caller of the plans verb. func (i *Inventory) PlanByID(ctx context.Context, id string) (Plan, error) { + if c, on, err := i.ChainReads(ctx, KindRollouts); err != nil { + return Plan{}, err + } else if on { + return i.chainPlanByID(ctx, c, id) + } plans, err := i.plans(ctx, `where id = $1`, id) if err != nil { return Plan{}, err diff --git a/internal/inventory/reachable.go b/internal/inventory/reachable.go index 5f53f937..41a96f45 100644 --- a/internal/inventory/reachable.go +++ b/internal/inventory/reachable.go @@ -1,11 +1,16 @@ package inventory import ( + "context" "errors" "io" "net" "github.com/jackc/pgx/v5/pgconn" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/chain" ) // Unreachable says whether an error means the store could not be asked right now, rather than @@ -29,6 +34,13 @@ func Unreachable(err error) bool { if err == nil { return false } + // The chain on the bus (novox/hq ADR 0408): a bus that does not answer, a lease this controller no longer holds, a key + // somebody else wrote first — each is "not now", as the store being away is. + if errors.Is(err, chain.ErrNotHeld) || errors.Is(err, chain.ErrMoved) || errors.Is(err, context.DeadlineExceeded) || + errors.Is(err, nats.ErrTimeout) || errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrConnectionClosed) || + errors.Is(err, nats.ErrDisconnected) || errors.Is(err, nats.ErrNoServers) || errors.Is(err, jetstream.ErrNoStreamResponse) { + return true + } var pg *pgconn.PgError if errors.As(err, &pg) { switch pg.Code { diff --git a/internal/link/source_moved_held_test.go b/internal/link/source_moved_held_test.go new file mode 100644 index 00000000..198e6682 --- /dev/null +++ b/internal/link/source_moved_held_test.go @@ -0,0 +1,60 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "io" + "log" + "testing" + "time" +) + +// heldControl is a merge announcement that records how it was settled. +type heldControl struct { + body []byte + took, held, drops int +} + +func (c *heldControl) Kind() string { return "source-moved" } +func (c *heldControl) Body() []byte { return c.body } +func (c *heldControl) Subject() string { return "" } +func (c *heldControl) Redelivered() bool { return false } +func (c *heldControl) HeldFor() time.Duration { return 0 } +func (c *heldControl) Answer(context.Context, []byte) error { return nil } +func (c *heldControl) Took() error { c.took++; return nil } +func (c *heldControl) Hold(time.Duration) error { c.held++; return nil } +func (c *heldControl) Drop() error { c.drops++; return nil } + +type sourceMover struct { + Upgrader + answer error +} + +func (s sourceMover) SourceMoved(context.Context, SourceMoved) error { return s.answer } + +// **A merge is acknowledged only after the controller recorded it** (novox/hq ADR 0407 decision 3, issue 1092): an +// answer of "not now" — the plan it belongs to could not be saved, the bus away, the lease lost — leaves the announcement +// unacknowledged, held to be handed over again; only an answer that is not "not now" settles it. +func TestAMergeTheControllerCouldNotRecordIsHeldAndNotAcknowledged(t *testing.T) { + body, err := json.Marshal(SourceMoved{Owner: "novox", Repo: "mesh-catalog", Base: "main", Commit: "c1"}) + if err != nil { + t.Fatal(err) + } + s := &Server{log: log.New(io.Discard, "", 0)} + for name, tc := range map[string]struct { + answer error + took, held int + }{ + "recorded": {nil, 1, 0}, + "not now": {errors.Join(ErrTryAgain, errors.New("the plan is not saved")), 0, 1}, + "an answer, not a wait": {errors.New("the mesh could not act on it"), 1, 0}, + } { + s.upgrader = sourceMover{answer: tc.answer} + m := &heldControl{body: body} + s.sourceMoved(context.Background(), m) + if m.took != tc.took || m.held != tc.held || m.drops != 0 { + t.Errorf("%s: acknowledged %d times and held %d times, want %d and %d", name, m.took, m.held, tc.took, tc.held) + } + } +} diff --git a/testdata/store-tests b/testdata/store-tests index 52dcabd7..49d5da6a 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -147,3 +147,5 @@ ./internal/inventory TestBuildsOnTheBusAreReadBackByEveryReader ./internal/inventory TestARequestIsTheBuildInStateAskedAndItsOutcomeTakesItsPlace ./internal/inventory TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain +./internal/inventory TestTheMoveCarriesThePlansAndTheirMergesAndTheyReadTheSame +./cmd/mesh-controller TestAMergeWhosePlanCannotBeSavedIsHeldAndRecordedWhenHandedOverAgain