diff --git a/cmd/mesh-controller/build.go b/cmd/mesh-controller/build.go index f4c78b0d..310c9d91 100644 --- a/cmd/mesh-controller/build.go +++ b/cmd/mesh-controller/build.go @@ -273,4 +273,7 @@ type answers struct { // applied, is not well: an assignment somebody made is not made yet, or will not be. pending []inventory.PendingAssignment pendingUnread string + // takeIn is the take-in's backlog of done keys on the bus (novox/hq ADR 0544 §6), nil where it cannot be read + // (a process with no bus, or a consumer not made yet). + takeIn *takeInBacklog } diff --git a/cmd/mesh-controller/composed_integration_test.go b/cmd/mesh-controller/composed_integration_test.go index 5c87f775..28d7cc41 100644 --- a/cmd/mesh-controller/composed_integration_test.go +++ b/cmd/mesh-controller/composed_integration_test.go @@ -102,8 +102,10 @@ func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) { // The new node-engine registered at its fingerprint, from its done key. n6 := fpOf('6') keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab") - if _, err := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil { - t.Fatal(err) + settled, _ := liveTakeInSteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}, m.js).batch(ctx, + []delivered{{module: "mesh-host", again: true}}, map[string]string{}) + if len(settled) != 1 || settled[0].err != nil || settled[0].verdict != acknowledged { + t.Fatalf("the take-in of mesh-host: %+v", settled) } if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil { t.Fatal(err) diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 2b2ac625..70c269b4 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -4,6 +4,8 @@ import ( "context" "errors" "fmt" + "maps" + "slices" "sort" "strings" "sync" @@ -12,210 +14,246 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/builderkeys" "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" ) // A new fingerprint of a registered module, from its done key to its compositions (novox/hq ADR 0539 clauses 5 and 7, -// ADR 0501 §2). With no delivery plan left to take a module's done key in, a write of the builder's or the indexer's -// keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names (registers the -// module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what a module's -// new manifest needs, a credential among it, as a send made it), composes the node holding the bus (its user list now -// carries the module's permissions at the new fingerprint: a new revision of the bus module's composition alone, -// written into its want at once by the assigner), and writes `grants..`. The assigner's sixth and seventh -// rules then let a want name the fingerprint. **No timer**: woken by the keys, and a starting controller looks at every -// module once (doneKeysAtStart). +// ADR 0501 §2, ADR 0544). With no delivery plan left to take a module's done key in, a write of the builder's or the +// indexer's keys does it: the controller takes in the done key of the fingerprint the indexer's trunk value names +// (registers the module at it, as a plan's take-in did), composes every node the module is assigned to (which makes what +// a module's new manifest needs, a credential among it), composes the node holding the bus (its user list now carries +// the module's permissions at the new fingerprint), and writes `grants..`. The assigner's sixth and seventh +// rules then let a want name the fingerprint. +// +// **A durable pull consumer, not a watch** (ADR 0544; research 072, milestones 2 and 4). The writes are read from two +// durable consumers (take_in_consumer.go): the bus keeps what is not taken in yet, so a restart takes in only what the +// controller before it left unacknowledged, and the start's look is the consumers' own catch-up. A batch of keys asks the +// composer twice whatever its size, and a key with nothing new composes nothing. -// doneKeySteps are what taking a module's done key in does, each a step a test may stand in for. -type doneKeySteps struct { - // held says the catalogue holds the module (a new one is the new-module take-in's, new_modules.go). - held func(ctx context.Context, module string) (bool, error) +// takeInSteps are what taking a batch of done keys in does, each a step a test may stand in for. +type takeInSteps struct { + // catalogue is every module the catalogue holds, with the nodes it is assigned to: read once a batch. A module it + // does not hold is the new-module take-in's (new_modules.go). + catalogue func(ctx context.Context) (map[string][]string, error) // trunk is the indexer's trunk value of the module. trunk func(ctx context.Context, module string) (sdkdelivery.Module, error) // registeredAt says the module is registered at the commit. registeredAt func(ctx context.Context, module, commit string) (bool, error) - // takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey); errKeyWaits when - // the key is not done yet. + // takeIn registers the module from the builder's done key of the trunk's fingerprint (takeInKey). takeIn func(ctx context.Context, module string, trunk sdkdelivery.Module) error - // nodesOf are the nodes the module is assigned to; busHolder the node holding the bus, "" for none. - nodesOf func(ctx context.Context, module string) ([]string, error) + // taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose. + taken func(ctx context.Context, module, fp string, nodes []string) (bool, error) + // busHolder is the node holding the bus, "" for none. busHolder func(ctx context.Context) (string, error) - // compose writes the nodes' compositions; grants writes the grants issued. - compose func(ctx context.Context, nodes []string) error - grants func(ctx context.Context) error + // compose composes the nodes, and writes the grants after when grants: each node's error, and the grants'. + compose func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) } -// take takes module's done key in and composes what it changed. It answers what it did in a line, "" when there was -// nothing to do. -func (s doneKeySteps) take(ctx context.Context, module string) (string, error) { - held, err := s.held(ctx, module) - if err != nil || !held { - return "", err - } - trunk, err := s.trunk(ctx, module) - if errors.Is(err, builderkeys.ErrNoValue) { - return "", nil - } - if err != nil { - return "", err - } - if !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2) { - return "", nil - } - at, err := s.registeredAt(ctx, module, trunk.Commit) - if err != nil { - return "", err - } - said := "" - if !at { - if err := s.takeIn(ctx, module, trunk); err != nil { - return "", err - } - said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", module, short(trunk.FP2), short(trunk.Commit)) - } - nodes, err := s.nodesOf(ctx, module) - if err != nil { - return said, err - } - if len(nodes) == 0 { - return said, nil - } - // The module's nodes first: composing them makes what the new manifest needs, its credentials among it. - if err := s.compose(ctx, nodes); err != nil { - return said, err - } - // Then the bus's user list, which carries the module's permissions at its new fingerprint. - holder, err := s.busHolder(ctx) - if err != nil { - return said, err - } - if holder != "" && !inList(nodes, holder) { - if err := s.compose(ctx, []string{holder}); err != nil { - return said, err - } - } - // Then the grant, which with the composition lets a want name the fingerprint (rules 6 and 7). - if err := s.grants(ctx); err != nil { - return said, err - } - if said == "" { - said = fmt.Sprintf("%s at %s", module, short(trunk.FP2)) - } - return said + "; composed for " + strings.Join(nodes, ", ") + ", its grant written", nil +// delivered is one key a consumer delivered: its module, and whether it was delivered before (a take-in that may have +// stopped half-way, so it is taken in whole). +type delivered struct { + module string + again bool } -func inList(list []string, s string) bool { - for _, v := range list { - if v == s { - return true - } - } - return false -} +// verdict is what becomes of a module's delivered keys. +type verdict int -// doneKeyTakeOnce is one take-in; a variable so a test may stand in for it. doneKeysSay says a line of the take-ins. -var ( - doneKeyTakeOnce = takeDoneKeyOnce - doneKeysSay = func(s string) { fmt.Println(s) } +const ( + // acknowledged: taken in, nothing new, a wait a later write ends, or a build key that failed. + acknowledged verdict = iota + // refused: delivered again by the bus after doneKeyRetry (a wait on a read, or a failure). + refused ) -// doneKeysBeforeLetGo is called just before a follower lets go of its module; a variable so a test can wake the module -// at that moment. Nil does nothing. -var doneKeysBeforeLetGo func(module string) - -// doneKeysLookAgain marks a module woken while a take-in of it ran: that take-in goes round once more. -var doneKeysLookAgain sync.Map - -// doneKeysMu keeps one take-in of a module's done key at a time in this process, per module. -var doneKeysMu sync.Map - -// takeDoneKey is a woken module's take-in on the live stores and buckets, said. A take-in that fails (its compositions -// or its grant not written) is tried again every doneKeyRetry while it fails, loudly: printed each time, and an urgent -// condition raised by the first until it has worked, as the chain's own moves are (layBuildsOutSoon). A key not done -// yet is no failure: it is said once for each reason it waits for and looked at again every doneKeyRetry. -func takeDoneKey(ctx context.Context, open *stores, module string) { - mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{}) - for { - if !mu.(*sync.Mutex).TryLock() { - // A take-in of the module already follows it: it is told to look again once it is done, so a newer key - // written meanwhile is never lost (the review of issue 1945). - doneKeysLookAgain.Store(module, true) - return - } - followDoneKey(ctx, open, module) - if doneKeysBeforeLetGo != nil { - doneKeysBeforeLetGo(module) - } - mu.(*sync.Mutex).Unlock() - // A wake between this follower's last look and its letting go failed its TryLock and only marked the module: - // looked at again now that it is let go, and taken once more (the review of issue 1945). - if _, again := doneKeysLookAgain.LoadAndDelete(module); !again || ctx.Err() != nil { - return - } - } +// settledKey is what a batch did with one module. +type settledKey struct { + module string + verdict verdict + // taken: registered, composed and granted now (not nothing new); its condition is cleared. + taken bool + // nothingNew: registered and composed before, so nothing was done. + nothingNew bool + // waits is why it waits, "" when it does not. + waits string + // err is a failure: raised as the module's condition. + err error + // said is the line said of it, "" for none. + said string } -// followDoneKey takes a module's done key in, under the module's lock, until it is taken in or ctx ends: a failure or a -// wait is tried again every doneKeyRetry. -func followDoneKey(ctx context.Context, open *stores, module string) { - failed := false - saidWait := "" - for { - doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it - err := takeOnceInSlot(ctx, open, module) - o := doneKeyObservation(module) - if err == nil { - // Cleared on every success, whoever raised it: a controller before a restart or another follower (issue - // 1945). A condition that is not open costs one read. - if keeper := conditionsFrom; keeper != nil { - _, _ = keeper.Clear(ctx, o.Key(), "composed and granted") - } - failed = false - if _, again := doneKeysLookAgain.LoadAndDelete(module); again { - continue // a wake came while this one ran: once more, for the key written meanwhile - } - return +// batch takes the delivered keys in, each module once: it answers one settledKey per module in the order first +// delivered, and the number of composer passes it asked. memo is, per module, the trunk (fingerprint and commit) this +// process took in last; a module at the same trunk delivered for the first time has nothing new. +func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[string]string) ([]settledKey, int) { + var order []string + again := map[string]bool{} + for _, k := range keys { + if _, seen := again[k.module]; !seen { + order = append(order, k.module) } - if errors.Is(err, errKeyWaits) { - // A take-in that waits is said once for each reason, and tried again every doneKeyRetry: the key write that - // would wake it may never come (issue 1945, live on 2026-10-25: mesh-host's take-in waited unsaid). - if why := err.Error(); why != saidWait { - doneKeysSay(fmt.Sprintf("done keys: %s waits: %s; looked at again every %s", module, - strings.TrimPrefix(why, errKeyWaits.Error()+": "), doneKeyRetry)) - saidWait = why - } - } else { - doneKeysSay(fmt.Sprintf("done keys: %s: %v; tried again in %s", module, err, doneKeyRetry)) + again[k.module] = again[k.module] || k.again + } + out := make([]settledKey, 0, len(order)) + all := func(v verdict, err error) ([]settledKey, int) { + for _, m := range order { + out = append(out, settledKey{module: m, verdict: v, err: err}) } - if errors.Is(err, errKeyWaits) { - select { - case <-ctx.Done(): - return - case <-time.After(doneKeyRetry): - } + return out, 0 + } + held, err := s.catalogue(ctx) + if err != nil { + return all(refused, fmt.Errorf("the catalogue cannot be read: %w", err)) + } + type toCompose struct { + at int + nodes []string + mark string + said string + } + var need []toCompose + for _, m := range order { + r := settledKey{module: m} + nodes, isHeld := held[m] + if !isHeld { + out = append(out, r) continue } - if ctx.Err() != nil { - return // stopped: a take-in cut by the stop is no fault of the build, and the next controller takes it in + trunk, err := s.trunk(ctx, m) + if errors.Is(err, builderkeys.ErrNoValue) || (err == nil && !strings.HasPrefix(trunk.FP2, sdkdelivery.SourcePrefix2)) { + out = append(out, r) + continue } - if keeper := conditionsFrom; keeper != nil && !failed && !errors.Is(err, context.Canceled) { - // A take-in whose own step was cancelled is tried again, and raises nothing: a cancellation is no fault - // of the build (issue 1945: 30 urgent conditions live, each a take-in cut by a restart). - o.Summary += err.Error() - if _, oerr := keeper.Observe(ctx, o); oerr == nil { - failed = true + if err != nil { + r.verdict, r.err = refused, err + out = append(out, r) + continue + } + mark := trunk.FP2 + "@" + trunk.Commit + if !again[m] && memo[m] == mark { + r.nothingNew = true + out = append(out, r) + continue + } + at, err := s.registeredAt(ctx, m, trunk.Commit) + if err != nil { + r.verdict, r.err = refused, err + out = append(out, r) + continue + } + said := "" + if !at { + if err := s.takeIn(ctx, m, trunk); err != nil { + switch { + case errors.Is(err, errWaitsForAWrite): + r.waits = err.Error() + case errors.Is(err, errKeyWaits): + r.verdict, r.waits = refused, err.Error() + case errors.Is(err, errKeyFailed): + r.err = err // acknowledged: only a new build ends it, and its key is delivered then + default: + r.verdict, r.err = refused, err + } + out = append(out, r) + continue + } + said = fmt.Sprintf("%s registered from the builder's done key %s (%s)", m, short(trunk.FP2), short(trunk.Commit)) + } else if !again[m] { + // Registered at its trunk already: composed and granted by a take-in before this one, or at this + // controller's start, unless that take-in stopped half-way, which leaves a composition or the grant missing. + if done, err := s.taken(ctx, m, trunk.FP2, nodes); err == nil && done { + memo[m] = mark + r.nothingNew = true + out = append(out, r) + continue } } - select { - case <-ctx.Done(): - return - case <-time.After(doneKeyRetry): + if len(nodes) == 0 { + memo[m] = mark + r.taken, r.said = true, said + out = append(out, r) + continue + } + if said == "" { + said = fmt.Sprintf("%s at %s", m, short(trunk.FP2)) + } + need = append(need, toCompose{at: len(out), nodes: nodes, mark: mark, said: said}) + out = append(out, r) + } + if len(need) == 0 { + return out, 0 + } + refuseNeed := func(err error) ([]settledKey, int) { + for _, n := range need { + out[n.at].verdict, out[n.at].err = refused, err + } + return out, 0 + } + holder, err := s.busHolder(ctx) + if err != nil { + return refuseNeed(fmt.Errorf("the node holding the bus cannot be read: %w", err)) + } + // The modules' nodes first, in one pass: composing them makes what their new manifests need, credentials among it. + var union []string + for _, n := range need { + for _, node := range n.nodes { + if node != holder && !slices.Contains(union, node) { + union = append(union, node) + } } } + sort.Strings(union) + passes := 0 + failed := map[string]error{} + if len(union) > 0 { + f, _, err := s.compose(ctx, union, false) + passes++ + if err != nil { + return refuseNeed(err) + } + maps.Copy(failed, f) + } + // Then the bus's user list, which carries the modules' permissions at their new fingerprints, and the grants, which + // with the compositions let a want name each fingerprint (rules 6 and 7). + var bus []string + if holder != "" { + bus = []string{holder} + } + f, grantsErr, err := s.compose(ctx, bus, true) + passes++ + if err != nil { + return refuseNeed(err) + } + maps.Copy(failed, f) + for _, n := range need { + var errs []error + for _, node := range append(slices.Clone(n.nodes), bus...) { + if e := failed[node]; e != nil && !slices.ContainsFunc(errs, func(x error) bool { return x == e }) { + errs = append(errs, e) + } + } + if grantsErr != nil { + errs = append(errs, grantsErr) + } + r := &out[n.at] + if len(errs) > 0 { + r.verdict, r.err = refused, errors.Join(errs...) + continue + } + memo[r.module] = n.mark + r.taken = true + r.said = n.said + "; composed for " + strings.Join(n.nodes, ", ") + ", its grant written" + } + return out, passes } -// doneKeyRetry is how long a failed take-in of a done key waits before it tries again. +// doneKeysSay says a line of the take-ins; a variable so a test may hear it. +var doneKeysSay = func(s string) { fmt.Println(s) } + +// doneKeyRetry is how long the bus waits before it delivers again a key whose take-in was refused. var doneKeyRetry = 30 * time.Second // kindNotComposed is the condition of a module's new fingerprint whose compositions or grant are not written: until @@ -228,30 +266,20 @@ func doneKeyObservation(module string) conditions.Observation { Summary: module + "'s new fingerprint is not composed or granted, so no want can name it: "} } -// takeDoneKeyOnce is one take-in of a module's done key. -func takeDoneKeyOnce(ctx context.Context, open *stores, module string) error { - keys, closeKeys, err := openBuilderKeys() - if err != nil { - return fmt.Errorf("the builder's keys cannot be read: %w", err) - } - defer closeKeys() - said, err := liveDoneKeySteps(open, keys).take(ctx, module) - if err == nil && said != "" { - fmt.Printf("done keys: %s\n", said) - } - return err -} - -// liveDoneKeySteps are the steps on the controller's own stores and the bus. -func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps { +// liveTakeInSteps are the steps on the controller's own stores and the bus. +func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) takeInSteps { inv := open.inventory - return doneKeySteps{ - held: func(ctx context.Context, module string) (bool, error) { - _, err := inv.SourceOf(ctx, module) - if errors.Is(err, inventory.ErrNoSuchModule) { - return false, nil + return takeInSteps{ + catalogue: func(ctx context.Context) (map[string][]string, error) { + entries, err := inv.Catalogued(ctx) + if err != nil { + return nil, err } - return err == nil, err + out := make(map[string][]string, len(entries)) + for _, e := range entries { + out[e.Manifest.Module] = e.On + } + return out, nil }, trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) { return keys.ModuleAt(ctx, module, "") @@ -270,17 +298,8 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps { } return err }, - nodesOf: func(ctx context.Context, module string) ([]string, error) { - entries, err := inv.Catalogued(ctx) - if err != nil { - return nil, err - } - for _, e := range entries { - if e.Manifest.Module == module { - return e.On, nil - } - } - return nil, nil + taken: func(ctx context.Context, module, fp string, nodes []string) (bool, error) { + return composedAndGranted(ctx, js, module, fp, nodes) }, busHolder: func(ctx context.Context) (string, error) { holders, err := seatHolders(ctx, inv) @@ -291,123 +310,46 @@ func liveDoneKeySteps(open *stores, keys registerKeys) doneKeySteps { }, // The composer's, never the take-in's own: one pass at a time composes each node asked for once, and writes the // grants once (issue 1945, composer.go). - compose: func(ctx context.Context, nodes []string) error { - return composerFor(open).compose(ctx, nodes, false) - }, - grants: func(ctx context.Context) error { - return composerFor(open).compose(ctx, nil, true) + compose: func(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) { + return composerFor(open).composeEach(ctx, nodes, grants) }, } } -// watchDoneKeys wakes the take-in of a module's done key for each module the builder or the indexer writes, until ctx -// ends: never on the watch's own goroutine. -func watchDoneKeys(ctx context.Context, open *stores) { - keys, closeKeys, err := openBuilderKeys() +// composedAndGranted says the bus holds module's grant at fp and its composition at fp for every node: a few reads of +// keys, never a composition. +func composedAndGranted(ctx context.Context, js jetstream.JetStream, module, fp string, nodes []string) (bool, error) { + grants, err := js.KeyValue(ctx, broker.GrantsBucket) if err != nil { - fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) - return + return false, err } - if err := builderkeys.Watch(ctx, keys.JS, func(module string) { - inBackground(ctx, func() { takeDoneKey(ctx, open, module) }) - }); err != nil { - closeKeys() - fmt.Printf("done keys: the builder's keys are not watched: %v\n", err) - return - } - goServing(ctx, func() { - <-ctx.Done() - closeKeys() - }) - // And every module once now, after the watch is open so no write falls between the two. - inBackground(ctx, func() { - doneKeysAtStart(ctx, catalogedModules(open), func(module string) { takeDoneKeyAtStart(ctx, open, module) }) - }) -} - -// doneKeysAtStart takes in the done key of every catalogued module once, one after another: the start's look the watch of -// new writes cannot give (it hears updates only). A done key written while no controller served, or whose take-in the -// controller this one replaced was stopped in, is taken here; without it the module stays in the node's own composition -// with no credential until its keys are written again (issue 1945, live on 2026-10-25 at 02:34). A take-in with nothing -// to do does nothing; a catalogue that cannot be read is said. -func doneKeysAtStart(ctx context.Context, modules func(context.Context) ([]string, error), take func(module string)) { - names, err := modules(ctx) + key, err := sdkdelivery.ControllerGrantKey(module, fp) if err != nil { - fmt.Printf("done keys: the modules are not looked at once at the start: %v\n", err) - return + return false, err } - sort.Strings(names) - // Side by side, a few at a time: their compositions meet in the composer's passes, each node composed once per pass - // (issue 1945), so the start's look is not one module's composition after another. - sem := make(chan struct{}, startTakeIns) - var wg sync.WaitGroup - for _, m := range names { - if ctx.Err() != nil { - break + if _, err := grants.Get(ctx, key); err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil } - sem <- struct{}{} - wg.Add(1) - go func() { - defer func() { <-sem; wg.Done() }() - take(m) - }() + return false, err } - wg.Wait() -} - -// startTakeIns is how many take-ins the start's look runs at once, and how many run at once in all. -const startTakeIns = 16 - -// takeInSlots are the take-ins that may run at once in this process, the watch's and the start's look's together: a -// burst of done keys (about 120 at 16:07:36 on 2026-10-25) once started a take-in each, all at once, and their reads -// filled the store until it refused every connection (issues 1945 and 1277). -var takeInSlots = make(chan struct{}, startTakeIns) - -// takeOnceInSlot is one take-in once a slot is free; a take-in that waits for its next try holds none. -func takeOnceInSlot(ctx context.Context, open *stores, module string) error { - select { - case takeInSlots <- struct{}{}: - case <-ctx.Done(): - return ctx.Err() + digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket) + if err != nil { + return false, err } - defer func() { <-takeInSlots }() - return doneKeyTakeOnce(ctx, open, module) -} - -// takeDoneKeyAtStart is one take-in of the start's look, under the module's lock; one that fails is handed to takeDoneKey, -// which tries again every doneKeyRetry, loudly, so a failing module never holds up the look at the others. -func takeDoneKeyAtStart(ctx context.Context, open *stores, module string) { - mu, _ := doneKeysMu.LoadOrStore(module, &sync.Mutex{}) - if !mu.(*sync.Mutex).TryLock() { - // A take-in already follows the module: told to look again, it takes what the start would have (issue 1945). - doneKeysLookAgain.Store(module, true) - return - } - doneKeysLookAgain.Delete(module) // what this look reads covers every wake before it - err := takeOnceInSlot(ctx, open, module) - mu.(*sync.Mutex).Unlock() - // Read after letting go, as every holder of a module's lock does: a wake during this look only marked the module - // (the review of issue 1945). - _, again := doneKeysLookAgain.LoadAndDelete(module) - if err != nil || again { - // A failure, a wait or a wake meanwhile is followed by takeDoneKey, which says it and looks again. - inBackground(ctx, func() { takeDoneKey(ctx, open, module) }) - } -} - -// catalogedModules are the names of every module the catalogue holds. -func catalogedModules(open *stores) func(context.Context) ([]string, error) { - return func(ctx context.Context) ([]string, error) { - entries, err := open.inventory.Catalogued(ctx) + for _, node := range nodes { + key, err := sdkdelivery.ComposedKey(node, module, fp) if err != nil { - return nil, err + return false, err } - out := make([]string, 0, len(entries)) - for _, e := range entries { - out = append(out, e.Manifest.Module) + if _, err := digests.Get(ctx, key); err != nil { + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return false, err } - return out, nil } + return true, nil } // doneKeysComposers are the composers of the done keys' take-ins in this process, one for each set of stores (a test diff --git a/cmd/mesh-controller/composed_keys_cancel_test.go b/cmd/mesh-controller/composed_keys_cancel_test.go deleted file mode 100644 index 0e34904c..00000000 --- a/cmd/mesh-controller/composed_keys_cancel_test.go +++ /dev/null @@ -1,83 +0,0 @@ -package main - -import ( - "context" - "fmt" - "testing" - "time" - - "github.com/novox/mesh-controller/internal/conditions" -) - -// A take-in cut by a cancellation raises no condition (issue 1945: on 2026-10-25 thirty urgent fingerprint-not-composed -// conditions were live, each a take-in a stop or a restart cut with "context canceled"): when the controller stops, the -// next one takes it in; when a step of its own was cancelled, it is tried again. -func TestATakeInCutByACancellationRaisesNothing(t *testing.T) { - for _, tc := range []struct { - name string - stop bool // the controller's context ends, else only the step's error is a cancellation - }{{"the controller stops", true}, {"a step was cancelled", false}} { - t.Run(tc.name, func(t *testing.T) { - was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom - t.Cleanup(func() { - doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper - }) - store := conditions.NewInMemory() - k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, - Teller: &conditions.Told{}}) - t.Cleanup(func() { k.Close(context.Background()) }) - conditionsFrom = k - doneKeyRetry = time.Millisecond - doneKeysSay = func(string) {} - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - calls++ - if tc.stop { - cancel() // stopped mid-way - return fmt.Errorf("the module cannot be composed: %w", context.Canceled) - } - if calls == 1 { - return fmt.Errorf("the module cannot be composed: %w", context.Canceled) - } - return nil - } - done := make(chan struct{}) - go func() { takeDoneKey(ctx, nil, "cut-module"); close(done) }() - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("the take-in did not end") - } - if !tc.stop && calls != 2 { - t.Errorf("a take-in whose step was cancelled was tried %d time(s), want 2", calls) - } - if c, found, err := conditions.ReadOne(t.Context(), store, doneKeyObservation("cut-module").Key()); err != nil || found { - t.Fatalf("a take-in cut by a cancellation raised %+v (found %v, %v)", c, found, err) - } - }) - } -} - -// A take-in that succeeds clears the module's fingerprint-not-composed condition whoever raised it: a controller before a -// restart, or another follower (issue 1945: the operator read "a build needs a look" for take-ins that had since worked). -func TestASucceedingTakeInClearsAConditionAnotherRaised(t *testing.T) { - was, wasRetry, wasSay, wasKeeper := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay, conditionsFrom = was, wasRetry, wasSay, wasKeeper }) - store := conditions.NewInMemory() - k := conditions.NewKeeper(t.Context(), conditions.Options{Store: store, History: store, Teller: &conditions.Told{}}) - t.Cleanup(func() { k.Close(context.Background()) }) - conditionsFrom = k - doneKeysSay = func(string) {} - o := doneKeyObservation("raised-before") - o.Summary += "the store refused the connection" - if _, err := k.Observe(t.Context(), o); err != nil { - t.Fatal(err) - } - doneKeyTakeOnce = func(context.Context, *stores, string) error { return nil } - takeDoneKey(t.Context(), nil, "raised-before") - if c, found, err := conditions.ReadOne(t.Context(), store, o.Key()); err != nil || found { - t.Fatalf("a take-in that succeeded left the condition another raised: %+v (found %v, %v)", c, found, err) - } -} diff --git a/cmd/mesh-controller/composed_keys_cap_test.go b/cmd/mesh-controller/composed_keys_cap_test.go deleted file mode 100644 index 54303253..00000000 --- a/cmd/mesh-controller/composed_keys_cap_test.go +++ /dev/null @@ -1,58 +0,0 @@ -package main - -import ( - "context" - "fmt" - "sync" - "sync/atomic" - "testing" - "time" -) - -// A burst of done keys runs at most startTakeIns take-ins at once, the watch's and the start's look together: at -// 16:07:36 on 2026-10-25 the builder rewrote about 120 done keys at once, each woke its own take-in, and each opened -// pools of its own until novox's store refused every connection (issues 1945 and 1277). -func TestABurstOfDoneKeysRunsAtMostSixteenTakeInsAtOnce(t *testing.T) { - was := doneKeyTakeOnce - t.Cleanup(func() { doneKeyTakeOnce = was }) - var running, most atomic.Int64 - var taken atomic.Int64 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - now := running.Add(1) - for { - m := most.Load() - if now <= m || most.CompareAndSwap(m, now) { - break - } - } - time.Sleep(2 * time.Millisecond) - running.Add(-1) - taken.Add(1) - return nil - } - ctx := context.Background() - var wg sync.WaitGroup - // The watch's wakes: one take-in each, in the background, as watchDoneKeys starts them. - for i := range 120 { - wg.Add(1) - go func() { defer wg.Done(); takeDoneKey(ctx, nil, fmt.Sprintf("burst-%03d", i)) }() - } - // And the start's look at the same moment, over other modules. - wg.Add(1) - go func() { - defer wg.Done() - var start []string - for i := range 40 { - start = append(start, fmt.Sprintf("start-%03d", i)) - } - doneKeysAtStart(ctx, func(context.Context) ([]string, error) { return start, nil }, - func(m string) { takeDoneKeyAtStart(ctx, nil, m) }) - }() - wg.Wait() - if got := taken.Load(); got != 160 { - t.Fatalf("took in %d modules, want 160", got) - } - if m := most.Load(); m > startTakeIns { - t.Fatalf("ran %d take-ins at once, want at most %d", m, startTakeIns) - } -} diff --git a/cmd/mesh-controller/composed_keys_start_test.go b/cmd/mesh-controller/composed_keys_start_test.go deleted file mode 100644 index 5dece4d8..00000000 --- a/cmd/mesh-controller/composed_keys_start_test.go +++ /dev/null @@ -1,34 +0,0 @@ -package main - -import ( - "context" - "errors" - "reflect" - "sort" - "sync" - "testing" -) - -// A starting controller takes in the done key of every catalogued module once, beside the watch of new writes: a done key -// written while no controller served, or whose take-in the controller it replaced was stopped in, is otherwise never -// taken, and its module stays in the node's own composition with no credential (issue 1945, live on 2026-10-25 at 02:34: -// the controller replaced mid take-in left mesh-provisioner-client, gitea, postgres and many more unregistered at their -// new fingerprints, and the new one, watching updates only, never looked). -func TestAStartingControllerTakesEveryModulesDoneKeyOnce(t *testing.T) { - var took []string - var mu sync.Mutex - doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { - return []string{"mesh-provisioner-client", "gitea", "postgres"}, nil - }, func(module string) { mu.Lock(); took = append(took, module); mu.Unlock() }) - sort.Strings(took) - if want := []string{"gitea", "mesh-provisioner-client", "postgres"}; !reflect.DeepEqual(took, want) { - t.Fatalf("took %v, want %v", took, want) - } - // A catalogue that cannot be read takes nothing and is said, never fatal. - took = nil - doneKeysAtStart(context.Background(), func(context.Context) ([]string, error) { return nil, errors.New("down") }, - func(module string) { took = append(took, module) }) - if len(took) != 0 { - t.Fatalf("took %v from a catalogue that could not be read", took) - } -} diff --git a/cmd/mesh-controller/composed_keys_test.go b/cmd/mesh-controller/composed_keys_test.go index b336ba9f..3abccf8b 100644 --- a/cmd/mesh-controller/composed_keys_test.go +++ b/cmd/mesh-controller/composed_keys_test.go @@ -4,84 +4,245 @@ import ( "context" "encoding/json" "errors" + "fmt" + "slices" "strings" "testing" + "time" sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-controller/internal/builderkeys" ) -// stepsFor are doneKeySteps over a module registered at registered, whose trunk value names fp at commit, assigned to -// nodes, the bus held on holder; every act is written to acts in order. -func stepsFor(acts *[]string, registered, commit, fp string, nodes []string, holder string) doneKeySteps { - return doneKeySteps{ - held: func(_ context.Context, module string) (bool, error) { return module == "mesh-mail", nil }, - trunk: func(context.Context, string) (sdkdelivery.Module, error) { - if fp == "" { +// aTakeIn is a mesh for the take-in's steps: per module its registered commit, its trunk (commit, fingerprint), its +// nodes, whether its compositions and grant exist, and what its take-in answers; every act is written to acts. +type aTakeIn struct { + registered map[string]string + trunks map[string]string // module -> commit; no entry: no trunk value + nodes map[string][]string + taken map[string]bool + takeInErr map[string]error + composeErr map[string]error + grantsErr error + holder string + acts []string +} + +func (m *aTakeIn) steps() takeInSteps { + fp := func(module string) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(module[:1], 64) } + return takeInSteps{ + catalogue: func(context.Context) (map[string][]string, error) { + m.acts = append(m.acts, "catalogue") + out := map[string][]string{} + for k, v := range m.nodes { + out[k] = v + } + return out, nil + }, + trunk: func(_ context.Context, module string) (sdkdelivery.Module, error) { + commit, ok := m.trunks[module] + if !ok { return sdkdelivery.Module{}, builderkeys.ErrNoValue } - return sdkdelivery.Module{Commit: commit, FP2: fp}, nil + return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil + }, + registeredAt: func(_ context.Context, module, commit string) (bool, error) { + return m.registered[module] == commit, nil }, - registeredAt: func(_ context.Context, _ string, c string) (bool, error) { return c == registered, nil }, takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error { - *acts = append(*acts, "take in "+module+" at "+trunk.Commit) - return nil + m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit) + return m.takeInErr[module] }, - nodesOf: func(context.Context, string) ([]string, error) { return nodes, nil }, - busHolder: func(context.Context) (string, error) { return holder, nil }, - compose: func(_ context.Context, ns []string) error { - *acts = append(*acts, "compose "+strings.Join(ns, ",")) - return nil + taken: func(_ context.Context, module, _ string, _ []string) (bool, error) { return m.taken[module], nil }, + busHolder: func(context.Context) (string, error) { return m.holder, nil }, + compose: func(_ context.Context, nodes []string, grants bool) (map[string]error, error, error) { + act := "compose " + strings.Join(nodes, ",") + if grants { + act += " and grants" + } + m.acts = append(m.acts, act) + failed := map[string]error{} + for _, n := range nodes { + if e := m.composeErr[n]; e != nil { + failed[n] = e + } + } + if grants { + return failed, m.grantsErr, nil + } + return failed, nil, nil }, - grants: func(context.Context) error { *acts = append(*acts, "grants"); return nil }, } } -// 1945/AC3 (unit half): a done key of a new fingerprint of an assigned module is taken in, then the module's nodes are -// composed, then the node holding the bus (its user list), then the grants: the grant and the composition exist -// before the assigner's rules let a want name the fingerprint. A module registered at the trunk's commit already is -// composed and granted without a second take-in; one assigned nowhere, unregistered, or with no trunk value is left -// alone. -func TestADoneKeyIsTakenInComposedAndGranted(t *testing.T) { - fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64) - var acts []string - said, err := stepsFor(&acts, "old", "new", fp, []string{"ace", "g14"}, "novox").take(context.Background(), "mesh-mail") - if err != nil { - t.Fatal(err) +// A batch of done keys takes each module in once and asks the composer at most twice, whatever its size: once for the +// nodes of every module with something new, then once for the node holding the bus with the grants (ADR 0544 §5). A +// module registered at its trunk whose compositions and grant exist, or that this process took in at the same trunk, +// composes nothing (§3): the old take-in composed it three times over, every key and every start (issue 1945, 17:36 on +// 2026-10-25). A key delivered again is taken in whole. A wait for a write is acknowledged, a wait on a read and a +// failure refused, a failed build key acknowledged with its condition (§4). +func TestATakeInBatch(t *testing.T) { + waitsForAWrite := fmt.Errorf("%w: the key is building", errWaitsForAWrite) + waitsOnARead := fmt.Errorf("%w: the store is away", errKeyWaits) + keyFailed := fmt.Errorf("%w: the build failed", errKeyFailed) + type want struct { + verdict verdict + taken, nothingNew bool + waits, err bool } - if got := strings.Join(acts, "; "); got != "take in mesh-mail at new; compose ace,g14; compose novox; grants" { - t.Errorf("the acts were %q", got) + cases := []struct { + name string + mesh aTakeIn + keys []delivered + memo map[string]string + acts []string + passes int + want map[string]want + }{ + {name: "a new fingerprint: taken in, its nodes, then the bus's node and the grants", + mesh: aTakeIn{registered: map[string]string{"mail": "old"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace", "g14", "novox"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "take in mail at new", "compose ace,g14", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}}}, + {name: "registered, composed and granted: nothing composed (fails on the old take-in)", + mesh: aTakeIn{registered: map[string]string{"mail": "new", "nats": "n"}, trunks: map[string]string{"mail": "new", "nats": "n"}, + nodes: map[string][]string{"mail": {"ace"}, "nats": {"novox"}}, taken: map[string]bool{"mail": true, "nats": true}, + holder: "novox"}, + keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}}, + acts: []string{"catalogue"}, + want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}}, + {name: "taken in by this process at the same trunk: nothing new, not even a look at its keys", + mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + memo: map[string]string{"mail": sdkdelivery.SourcePrefix2 + strings.Repeat("m", 64) + "@new"}, + acts: []string{"catalogue"}, want: map[string]want{"mail": {nothingNew: true}}}, + {name: "delivered again: taken in whole, even when it looks done", + mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace"}}, taken: map[string]bool{"mail": true}, holder: "novox"}, + keys: []delivered{{module: "mail", again: true}}, + acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}}}, + {name: "a take-in that stopped half-way: composed", + mesh: aTakeIn{registered: map[string]string{"mail": "new"}, trunks: map[string]string{"mail": "new"}, + nodes: map[string][]string{"mail": {"ace"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "compose ace", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}}}, + {name: "three modules, one pass for their nodes and one for the bus", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2", "bus": "3"}, + nodes: map[string][]string{"mail": {"ace", "novox"}, "web": {"g14", "ace"}, "bus": {"novox"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}, {module: "web"}, {module: "bus"}}, + acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "take in bus at 3", "compose ace,g14", + "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {taken: true}, "web": {taken: true}, "bus": {taken: true}}}, + {name: "assigned nowhere, held by no catalogue, no trunk value: nothing composed", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "odd": "2"}, + nodes: map[string][]string{"mail": nil, "none": {"ace"}}, holder: "novox"}, + keys: []delivered{{module: "mail"}, {module: "odd"}, {module: "none"}}, + acts: []string{"catalogue", "take in mail at 1"}, + want: map[string]want{"mail": {taken: true}, "odd": {}, "none": {}}}, + {name: "a grant not written refuses the key, never swallowed", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1"}, + nodes: map[string][]string{"mail": {"ace"}}, holder: "novox", grantsErr: errors.New("the bucket mesh-controller_grants: not found")}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "take in mail at 1", "compose ace", "compose novox and grants"}, passes: 2, + want: map[string]want{"mail": {verdict: refused, err: true}}}, + {name: "a node that fails refuses only its modules", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"mail": "1", "web": "2"}, + nodes: map[string][]string{"mail": {"ace"}, "web": {"g14"}}, holder: "novox", + composeErr: map[string]error{"ace": errors.New("store refused")}}, + keys: []delivered{{module: "mail"}, {module: "web"}}, + acts: []string{"catalogue", "take in mail at 1", "take in web at 2", "compose ace,g14", "compose novox and grants"}, + passes: 2, want: map[string]want{"mail": {verdict: refused, err: true}, "web": {taken: true}}}, + {name: "waits and failures: a write's wait acknowledged, a read's refused, a failed key acknowledged", + mesh: aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2", "c": "3"}, + nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}, "c": {"ace"}}, holder: "novox", + takeInErr: map[string]error{"a": waitsForAWrite, "b": waitsOnARead, "c": keyFailed}}, + keys: []delivered{{module: "a"}, {module: "b"}, {module: "c"}}, + acts: []string{"catalogue", "take in a at 1", "take in b at 2", "take in c at 3"}, + want: map[string]want{"a": {waits: true}, "b": {verdict: refused, waits: true}, "c": {err: true}}}, } - if !strings.Contains(said, "registered from the builder's done key") { - t.Errorf("said %q", said) + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + memo := c.memo + if memo == nil { + memo = map[string]string{} + } + settled, passes := c.mesh.steps().batch(t.Context(), c.keys, memo) + if !slices.Equal(c.mesh.acts, c.acts) { + t.Fatalf("the acts were\n %q\nwant\n %q", c.mesh.acts, c.acts) + } + if passes != c.passes { + t.Fatalf("%d composer pass(es), want %d", passes, c.passes) + } + if len(settled) != len(c.want) { + t.Fatalf("settled %d module(s), want %d: %+v", len(settled), len(c.want), settled) + } + for _, r := range settled { + w := c.want[r.module] + got := want{verdict: r.verdict, taken: r.taken, nothingNew: r.nothingNew, waits: r.waits != "", err: r.err != nil} + if got != w { + t.Errorf("%s: %+v, want %+v (%v)", r.module, got, w, r.err) + } + if r.taken && memo[r.module] == "" { + t.Errorf("%s was taken in and not remembered", r.module) + } + } + }) } - acts = nil - if _, err := stepsFor(&acts, "new", "new", fp, []string{"novox"}, "novox").take(context.Background(), "mesh-mail"); err != nil { - t.Fatal(err) - } - if got := strings.Join(acts, "; "); got != "compose novox; grants" { - t.Errorf("a module registered at its trunk already: %q", got) - } + // And each delivered key is acknowledged or refused by its module's verdict, the refused with the retry's delay. + t.Run("acknowledged and refused", func(t *testing.T) { + mesh := aTakeIn{registered: map[string]string{}, trunks: map[string]string{"a": "1", "b": "2"}, + nodes: map[string][]string{"a": {"ace"}, "b": {"ace"}}, takeInErr: map[string]error{"b": waitsOnARead}} + var acked, naked []string + msg := func(module string) takeInMsg { + return takeInMsg{delivered: delivered{module: module}, ack: func() { acked = append(acked, module) }, + nak: func(d time.Duration) { + if d != doneKeyRetry { + t.Errorf("refused with a delay of %s", d) + } + naked = append(naked, module) + }} + } + state := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}} + said := state.take(t.Context(), mesh.steps(), []takeInMsg{msg("a"), msg("b"), msg("a")}) + if !slices.Equal(acked, []string{"a", "a"}) || !slices.Equal(naked, []string{"b"}) { + t.Fatalf("acknowledged %v and refused %v", acked, naked) + } + if !strings.Contains(said, "1 taken in") || !strings.Contains(said, "1 waiting") { + t.Fatalf("said %q", said) + } + }) +} - for name, s := range map[string]doneKeySteps{ - "assigned nowhere": stepsFor(&acts, "old", "new", fp, nil, "novox"), - "no trunk value": stepsFor(&acts, "old", "new", "", []string{"ace"}, "novox"), - } { - acts = nil - if _, err := s.take(context.Background(), "mesh-mail"); err != nil { - t.Fatal(err) - } - if name == "assigned nowhere" && strings.Join(acts, "; ") != "take in mesh-mail at new" { - t.Errorf("%s: %v", name, acts) - } - if name == "no trunk value" && len(acts) != 0 { - t.Errorf("%s: %v", name, acts) - } +// A backlog that stands above zero with nothing delivered for five minutes is a stall; one that moves, or empties, is +// not (ADR 0544 §6). +func TestATakeInStallsWhenItsBacklogStandsStill(t *testing.T) { + at := time.Date(2026, 10, 25, 17, 36, 0, 0, time.UTC) + w := &stallWatch{} + read := func(pending uint64, delivered ...uint64) takeInBacklog { + return takeInBacklog{Pending: pending, delivered: delivered} } - acts = nil - if _, err := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox").take(context.Background(), "unknown"); err != nil || len(acts) != 0 { - t.Errorf("a module the catalogue does not hold: %v %v", acts, err) + steps := []struct { + after time.Duration + backlog takeInBacklog + stalled, changed bool + }{ + {0, read(190, 10, 3), false, false}, + {4 * time.Minute, read(190, 10, 3), false, false}, + {5 * time.Minute, read(190, 10, 3), true, true}, + {6 * time.Minute, read(190, 10, 3), true, false}, + {7 * time.Minute, read(180, 11, 3), false, true}, + {20 * time.Minute, read(0, 11, 3), false, false}, + } + for i, s := range steps { + stalled, changed := w.look(s.backlog, at.Add(s.after)) + if stalled != s.stalled || changed != s.changed { + t.Fatalf("reading %d: stalled %v changed %v, want %v %v", i, stalled, changed, s.stalled, s.changed) + } } } @@ -115,15 +276,3 @@ func TestAChangedUserListIsANewRevisionOfTheBusModulesCompositionAlone(t *testin } } } - -// 1945 (the review's D1): a grant that is not written is the take-in's error, never swallowed: with no send left, -// nothing else would write it, and no want could name the fingerprint. -func TestAGrantNotWrittenIsTheTakeInsError(t *testing.T) { - fp := sdkdelivery.SourcePrefix2 + strings.Repeat("c", 64) - var acts []string - s := stepsFor(&acts, "old", "new", fp, []string{"ace"}, "novox") - s.grants = func(context.Context) error { return errors.New("the bucket mesh-controller_grants: not found") } - if _, err := s.take(context.Background(), "mesh-mail"); err == nil || !strings.Contains(err.Error(), "grants") { - t.Errorf("a grant not written: %v", err) - } -} diff --git a/cmd/mesh-controller/composed_keys_waits_test.go b/cmd/mesh-controller/composed_keys_waits_test.go deleted file mode 100644 index 839f9b4f..00000000 --- a/cmd/mesh-controller/composed_keys_waits_test.go +++ /dev/null @@ -1,184 +0,0 @@ -package main - -import ( - "context" - "fmt" - "strings" - "sync" - "testing" - "time" -) - -// A take-in that waits is said once for each reason it waits for, and tried again every doneKeyRetry until it is taken -// in: never left for a key write that may not come (issue 1945, live on 2026-10-25: mesh-host's take-in at 26bd waited, -// said nothing, and nothing woke it until another write). -func TestAWaitingTakeInIsSaidOnceAndTriedAgain(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 5 * time.Millisecond - var mu sync.Mutex - var said []string - doneKeysSay = func(s string) { mu.Lock(); said = append(said, s); mu.Unlock() } - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - calls++ - switch { - case calls <= 2: - return fmt.Errorf("%w: the artifact store could not hold mesh-host's artifact", errKeyWaits) - case calls == 3: - return fmt.Errorf("%w: mesh-host's key cannot be read", errKeyWaits) - } - return nil - } - done := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(done) }() - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("a waiting take-in was not tried again until it was taken in") - } - if calls != 4 { - t.Fatalf("tried %d times, want 4", calls) - } - mu.Lock() - defer mu.Unlock() - waits := 0 - for _, s := range said { - if strings.Contains(s, "mesh-host waits") { - waits++ - } - } - if waits != 2 { - t.Fatalf("said %q: want each of the two reasons once", said) - } -} - -// A wake while a take-in of the module already follows it adds no second one: the follower looks again itself. -func TestAWakeWhileATakeInFollowsAddsNone(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 20 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - release := make(chan struct{}) - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - mu.Unlock() - select { - case <-release: - return nil - default: - return fmt.Errorf("%w: the key says building", errKeyWaits) - } - } - follower := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "board"); close(follower) }() - time.Sleep(5 * time.Millisecond) - returned := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "board"); close(returned) }() - select { - case <-returned: - case <-time.After(time.Second): - t.Fatal("a second wake waited for the follower") - } - close(release) - <-follower -} - -// A wake while a take-in of the module follows it is not lost: the follower goes round once more after it succeeds, so -// a newer build's key written meanwhile is taken in (the review of issue 1945). -func TestAWakeDuringATakeInIsTakenAfterIt(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 20 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - inFirst := make(chan struct{}) - release := make(chan struct{}) - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - n := calls - mu.Unlock() - if n == 1 { - close(inFirst) - <-release - } - return nil - } - follower := make(chan struct{}) - go func() { takeDoneKey(context.Background(), nil, "mesh-host"); close(follower) }() - <-inFirst - takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, while the first runs - close(release) - <-follower - mu.Lock() - defer mu.Unlock() - if calls != 2 { - t.Fatalf("taken %d times: the wake during the take-in was lost", calls) - } -} - -// A wake that comes after the follower's last look but before it lets go of the module is not lost: the follower looks -// at the mark again once it has let go, and takes the module once more (the review of issue 1945). -func TestAWakeJustBeforeTheFollowerLetsGoIsTaken(t *testing.T) { - was, wasRetry, wasSay, wasHook := doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo - t.Cleanup(func() { - doneKeyTakeOnce, doneKeyRetry, doneKeysSay, doneKeysBeforeLetGo = was, wasRetry, wasSay, wasHook - }) - doneKeyRetry = 10 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - mu.Unlock() - return nil - } - woke := false - doneKeysBeforeLetGo = func(module string) { - if !woke { - woke = true - takeDoneKey(context.Background(), nil, module) // fails its TryLock and marks the module - } - } - takeDoneKey(context.Background(), nil, "mesh-host") - mu.Lock() - defer mu.Unlock() - if calls != 2 { - t.Fatalf("taken %d times: the wake just before the follower let go was lost", calls) - } -} - -// A wake during the start's take-in of a module is not lost: the start pass reads the mark once it lets go of the module, -// and hands the module to takeDoneKey, which takes it once more (the review of issue 1945: the third dropped wake of its -// kind). -func TestAWakeDuringTheStartsTakeInIsTaken(t *testing.T) { - was, wasRetry, wasSay := doneKeyTakeOnce, doneKeyRetry, doneKeysSay - t.Cleanup(func() { doneKeyTakeOnce, doneKeyRetry, doneKeysSay = was, wasRetry, wasSay }) - doneKeyRetry = 10 * time.Millisecond - doneKeysSay = func(string) {} - var mu sync.Mutex - calls := 0 - doneKeyTakeOnce = func(context.Context, *stores, string) error { - mu.Lock() - calls++ - n := calls - mu.Unlock() - if n == 1 { - takeDoneKey(context.Background(), nil, "mesh-host") // the newer key's wake, during the start's take-in - } - return nil - } - takeDoneKeyAtStart(context.Background(), nil, "mesh-host") - background.Wait() - mu.Lock() - defer mu.Unlock() - if calls != 2 { - t.Fatalf("taken %d times: the wake during the start's take-in was lost", calls) - } -} diff --git a/cmd/mesh-controller/composer.go b/cmd/mesh-controller/composer.go index c6637eea..025321cb 100644 --- a/cmd/mesh-controller/composer.go +++ b/cmd/mesh-controller/composer.go @@ -29,7 +29,13 @@ type composer struct { type waiter struct { nodes []string grants bool - done chan error + done chan passResult +} + +// passResult is a waiter's answer: the error of each of its nodes that failed, and of the grants when it asked for them. +type passResult struct { + failed map[string]error + grants error } func newComposer(compose func(context.Context, []string) map[string]error, grants func(context.Context) error) *composer { @@ -39,7 +45,23 @@ func newComposer(compose func(context.Context, []string) map[string]error, grant // compose asks for the nodes to be composed, and the grants written after when grants is true, and waits for the pass // that does it, with the errors of its own nodes alone. func (c *composer) compose(ctx context.Context, nodes []string, grants bool) error { - done := make(chan error, 1) + failed, grantsErr, err := c.composeEach(ctx, nodes, grants) + if err != nil { + return err + } + var errs []error + for _, n := range nodes { + if e := failed[n]; e != nil { + errs = append(errs, e) + } + } + return errors.Join(append(errs, grantsErr)...) +} + +// composeEach is compose answering each node's error apart, and the grants' (ADR 0544 §5: a batch of take-ins asks one +// pass and settles each module by its own nodes); err is ctx's when it ends first. +func (c *composer) composeEach(ctx context.Context, nodes []string, grants bool) (map[string]error, error, error) { + done := make(chan passResult, 1) c.mu.Lock() for _, n := range nodes { c.pending[n] = true @@ -52,10 +74,10 @@ func (c *composer) compose(ctx context.Context, nodes []string, grants bool) err default: } select { - case err := <-done: - return err + case r := <-done: + return r.failed, r.grants, nil case <-ctx.Done(): - return ctx.Err() + return nil, nil, ctx.Err() } } @@ -90,16 +112,16 @@ func (c *composer) run(ctx context.Context) { grantsErr = c.writeGrants(ctx) } for _, w := range waiters { - var errs []error + r := passResult{failed: map[string]error{}} for _, n := range w.nodes { if err := failed[n]; err != nil { - errs = append(errs, err) + r.failed[n] = err } } - if w.grants && grantsErr != nil { - errs = append(errs, grantsErr) + if w.grants { + r.grants = grantsErr } - w.done <- errors.Join(errs...) + w.done <- r } } } diff --git a/cmd/mesh-controller/key_takein.go b/cmd/mesh-controller/key_takein.go index 3f00bbc2..abb3a88b 100644 --- a/cmd/mesh-controller/key_takein.go +++ b/cmd/mesh-controller/key_takein.go @@ -39,6 +39,11 @@ import ( // errKeyWaits says a take-in from a key waits: for the builder, a reference, the store or the forge. Never a failure. var errKeyWaits = errors.New("waits") +// errWaitsForAWrite is a take-in that waits for the builder or the indexer to write again (the key not done, its +// artifacts not referenced, no fingerprint of the new form): that write wakes it, so nothing looks again before it +// (ADR 0544 §4). +var errWaitsForAWrite = fmt.Errorf("%w for the builder's or the indexer's next write", errKeyWaits) + // errKeyFailed says the module's key failed: the builder's own words follow. var errKeyFailed = errors.New("the builder's key failed") @@ -298,12 +303,15 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) ( waits := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) { return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errKeyWaits}, a...)...) } + waitsForAWrite := func(format string, a ...any) (inventory.Build, catalogue.Manifest, error) { + return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: "+format, append([]any{errWaitsForAWrite}, a...)...) + } if t.Trunk.FP2 == "" && t.Hand == nil { why := t.Trunk.None2 if why == "" { why = "the indexer wrote no fingerprint of the new form" } - return waits("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why) + return waitsForAWrite("%s at %.8s has no fingerprint to take a build by: %s", t.Module, t.Trunk.Commit, why) } k, err := keys.Build(ctx, t.Module, t.Trunk.FP2) if t.Hand != nil { @@ -319,9 +327,9 @@ func buildOfKey(ctx context.Context, keys keyReads, t keyTake, kc keyContext) ( case k.Failed(): return inventory.Build{}, catalogue.Manifest{}, fmt.Errorf("%w: %s", errKeyFailed, k.Says()) case !k.Done(): - return waits("%s", k.Says()) + return waitsForAWrite("%s", k.Says()) case len(k.Unreferenced()) > 0: - return waits("%s", k.Says()) + return waitsForAWrite("%s", k.Says()) } src, err := keys.Manifest(ctx, k.Build.Manifest) if err != nil { diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 2b1fb1b0..1db4ca05 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -70,6 +70,13 @@ var plainWordings = map[string]func(conditions.Observation) words{ "none can be given it yet. It is tried again every half minute; the details say what fails.", m), Resolved: "the new build of " + m + " can be delivered"} }), + // The take-in of done keys whose backlog stood still (take_in_consumer.go, novox/hq ADR 0544 §6). + kindTakeInStalled: worded(func(o conditions.Observation) words { + return words{Headline: "New builds are not being prepared for the machines", + Explanation: "The mesh prepares each new build for the machines as it is made, and has prepared none for " + + "five minutes while builds wait. Until it moves, no new build reaches a machine; the details say how many wait.", + Resolved: "new builds are being prepared again"} + }), // Machines and their node-engine. "silent": worded(func(o conditions.Observation) words { m := machineOr(o, "a machine") diff --git a/cmd/mesh-controller/readable.go b/cmd/mesh-controller/readable.go index c71d9b21..3b6e53c1 100644 --- a/cmd/mesh-controller/readable.go +++ b/cmd/mesh-controller/readable.go @@ -111,6 +111,9 @@ type meshStatus struct { // why they could not be read. Pending []pendingStatus `json:"pending,omitempty"` PendingUnread string `json:"pendingUnread,omitempty"` + // TakeIn is the take-in's backlog of done keys: keys not delivered yet and keys being taken in (novox/hq ADR 0544 + // §6). Absent where it cannot be read. + TakeIn *takeInBacklog `json:"take-in,omitempty"` } // pendingStatus is one pending assignment as status says it. @@ -266,6 +269,7 @@ func statusAsJSON(asked answers) ([]byte, error) { out.Failing = providerStandings(asked.conditions) out.Overflowing = asked.overflowing out.PendingUnread = asked.pendingUnread + out.TakeIn = asked.takeIn for _, p := range asked.pending { out.Pending = append(out.Pending, pendingStatus{Node: p.Node, Module: p.Module, State: p.State, Build: p.Build, Repository: p.Repository, Path: p.Path, Since: p.Since, Settled: p.Settled, Note: p.Note}) diff --git a/cmd/mesh-controller/status.go b/cmd/mesh-controller/status.go index d2e940f2..a9f86272 100644 --- a/cmd/mesh-controller/status.go +++ b/cmd/mesh-controller/status.go @@ -151,6 +151,10 @@ func printStatus(asked answers) error { } printPending(asked.pending, asked.pendingUnread) + if b := asked.takeIn; b != nil && b.Total() > 0 { + fmt.Printf("the take-in of done keys has %d key(s) not taken in yet: %d waiting to be delivered, %d being taken in\n\n", + b.Total(), b.Pending, b.Taking) + } if len(behind) > 0 { var names []string @@ -471,6 +475,13 @@ func theThreeQuestions(ctx context.Context, open *stores) (answers, error) { out.pendingUnread = err.Error() err = nil } + // And the take-in's backlog of done keys, read from its consumers on the bus (novox/hq ADR 0544 §6): where there is + // a bus; a stall is a condition of its own, so this is said and judges nothing. + if _, onBus := broker.BusAddress(); onBus == nil { + if b, berr := readTakeInBacklog(ctx); berr == nil { + out.takeIn = &b + } + } // And how many repairs were done by hand this week (novox/hq to-be 45 §7) — where there is a bus // to read the log from; a process with none has no log to count. if _, onBus := broker.BusAddress(); onBus == nil { diff --git a/cmd/mesh-controller/take_in_consumer.go b/cmd/mesh-controller/take_in_consumer.go new file mode 100644 index 00000000..7592521b --- /dev/null +++ b/cmd/mesh-controller/take_in_consumer.go @@ -0,0 +1,365 @@ +package main + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "time" + + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/conditions" +) + +// The take-in's two durable pull consumers (novox/hq ADR 0544; research 072, milestones 2 and 4). On the indexer's +// modules bucket the consumer delivers each module's last trunk value when it is made, then each write: the start's look +// of a controller whose consumer is new, and nothing at a restart but what the controller before it left +// unacknowledged. On the builder's builds bucket it delivers each write from when it is made. A key is acknowledged once +// its module is taken in, has nothing new, or waits for a write that will be delivered; any other wait or failure is +// refused with a delay, and the bus delivers it again. At most startTakeIns keys of each are unacknowledged: the bound on +// the store (issues 1945 and 1277: a burst of take-ins once filled it until it refused every connection). + +// startTakeIns is how many keys each consumer leaves unacknowledged at most: the take-ins that may wait on the store. +const startTakeIns = 16 + +// takeInAckWait is how long a delivered key may stay unacknowledged before the bus delivers it again: longer than a batch +// takes, its two composer passes among it. +const takeInAckWait = 10 * time.Minute + +// takeInGather is how long a batch waits for more keys after its first, so a burst is one batch. +var takeInGather = 200 * time.Millisecond + +// takeInStall is how long a backlog may stand without a key delivered before it is a condition (ADR 0544 §6). +var takeInStall = 5 * time.Minute + +// takeInMsg is one delivered key, as the take-in uses it. +type takeInMsg struct { + delivered + ack func() + nak func(delay time.Duration) +} + +// takeInKeyOf is the module a delivered key names, false for a delete or a purge, which takes nothing in. +func takeInKeyOf(bucket string, m jetstream.Msg) (delivered, bool) { + if op := m.Headers().Get("KV-Operation"); op == "DEL" || op == "PURGE" { + return delivered{}, false + } + key := strings.TrimPrefix(m.Subject(), "$KV."+bucket+".") + module, _, _ := strings.Cut(key, ".") + if module == "" { + return delivered{}, false + } + again := false + if meta, err := m.Metadata(); err == nil { + again = meta.NumDelivered > 1 + } + return delivered{module: module, again: again}, true +} + +// takeInConsumer is one bucket's consumer of the take-in. +type takeInConsumer struct { + bucket string + consumer jetstream.Consumer +} + +// takeInConsumers makes or finds the take-in's consumer on each bucket's stream; one that cannot be made is said, and +// the other is still read. +func takeInConsumers(ctx context.Context, js jetstream.JetStream) []takeInConsumer { + var out []takeInConsumer + for _, bucket := range broker.TakeInBuckets { + deliver := jetstream.DeliverNewPolicy + if bucket == sdkdelivery.ModulesBucket { + deliver = jetstream.DeliverLastPerSubjectPolicy // the start's look: every module's trunk value once + } + c, err := js.CreateOrUpdateConsumer(ctx, "KV_"+bucket, jetstream.ConsumerConfig{ + Durable: broker.TakeInConsumer, FilterSubject: "$KV." + bucket + ".>", DeliverPolicy: deliver, + AckPolicy: jetstream.AckExplicitPolicy, AckWait: takeInAckWait, MaxAckPending: startTakeIns, + Description: "the controller's take-in of done keys (novox/hq ADR 0544)", + }) + if err != nil { + fmt.Printf("done keys: the take-in's consumer on %s is not made, so its keys are not taken in until the "+ + "next start: %v\n", bucket, err) + continue + } + out = append(out, takeInConsumer{bucket: bucket, consumer: c}) + } + return out +} + +// watchDoneKeys takes in the done keys the two consumers deliver, batch after batch, until ctx ends, and watches their +// backlog. A consumer that cannot be made is said: its bucket is then not taken in until the next start. +func watchDoneKeys(ctx context.Context, open *stores) { + keys, closeKeys, err := openBuilderKeys() + if err != nil { + fmt.Printf("done keys: the builder's keys are not taken in: %v\n", err) + return + } + consumers := takeInConsumers(ctx, keys.JS) + if len(consumers) == 0 { + closeKeys() + return + } + in := make(chan takeInMsg) + read := make([]jetstream.Consumer, 0, len(consumers)) + for _, c := range consumers { + bucket := c.bucket + read = append(read, c.consumer) + it, err := c.consumer.Messages(jetstream.PullMaxMessages(startTakeIns)) + if err != nil { + fmt.Printf("done keys: the take-in's consumer on %s is not read: %v\n", bucket, err) + continue + } + goServing(ctx, func() { + <-ctx.Done() + it.Stop() + }) + goServing(ctx, func() { + for { + m, err := it.Next() + if err != nil { + if ctx.Err() == nil && !errors.Is(err, jetstream.ErrMsgIteratorClosed) { + fmt.Printf("done keys: the take-in's consumer on %s stopped: %v\n", bucket, err) + } + return + } + d, ok := takeInKeyOf(bucket, m) + if !ok { + _ = m.Ack() + continue + } + msg := takeInMsg{delivered: d, ack: func() { _ = m.Ack() }, + nak: func(delay time.Duration) { _ = m.NakWithDelay(delay) }} + select { + case in <- msg: + case <-ctx.Done(): + return + } + } + }) + } + backlog := func(ctx context.Context) (takeInBacklog, error) { return backlogOf(ctx, read) } + steps := liveTakeInSteps(open, keys, keys.JS) + goServing(ctx, func() { takeInLoop(ctx, in, steps, backlog) }) + goServing(ctx, func() { watchTakeInStall(ctx, backlog) }) + goServing(ctx, func() { + <-ctx.Done() + closeKeys() + }) +} + +// takeInLoop gathers delivered keys into batches and takes each batch in, until ctx ends. +func takeInLoop(ctx context.Context, in <-chan takeInMsg, steps takeInSteps, + backlog func(context.Context) (takeInBacklog, error)) { + t := &takeInState{memo: map[string]string{}, raised: map[string]bool{}, saidWait: map[string]string{}} + for { + var batch []takeInMsg + select { + case <-ctx.Done(): + return + case m := <-in: + batch = append(batch, m) + } + gather := time.NewTimer(takeInGather) + more: + for len(batch) < 2*startTakeIns { + select { + case m := <-in: + batch = append(batch, m) + case <-gather.C: + break more + case <-ctx.Done(): + gather.Stop() + return + } + } + gather.Stop() + said := t.take(ctx, steps, batch) + if b, err := backlog(ctx); err == nil { + said += fmt.Sprintf("; backlog %d", b.Total()) + } + doneKeysSay("done keys: " + said) + } +} + +// takeInState is what the take-in keeps between batches in this process: the trunk each module was last taken in at, +// the modules whose condition it raised, and the wait last said of each. +type takeInState struct { + memo map[string]string + raised map[string]bool + saidWait map[string]string +} + +// take takes one batch in and acknowledges or refuses each key, raising and clearing the modules' conditions. It answers +// the batch in a line. +func (t *takeInState) take(ctx context.Context, steps takeInSteps, batch []takeInMsg) string { + keys := make([]delivered, len(batch)) + for i, m := range batch { + keys[i] = m.delivered + } + settled, passes := steps.batch(ctx, keys, t.memo) + verdicts := make(map[string]verdict, len(settled)) + var taken, nothing, waiting, failed int + for _, r := range settled { + verdicts[r.module] = r.verdict + o := doneKeyObservation(r.module) + switch { + case r.err != nil: + failed++ + doneKeysSay(fmt.Sprintf("done keys: %s: %v", r.module, r.err)) + if keeper := conditionsFrom; keeper != nil && !t.raised[r.module] && ctx.Err() == nil && + !errors.Is(r.err, context.Canceled) { + o.Summary += r.err.Error() + if _, err := keeper.Observe(ctx, o); err == nil { + t.raised[r.module] = true + } + } + case r.waits != "": + waiting++ + if t.saidWait[r.module] != r.waits { + doneKeysSay(fmt.Sprintf("done keys: %s %s", r.module, r.waits)) + t.saidWait[r.module] = r.waits + } + case r.taken: + taken++ + delete(t.saidWait, r.module) + if r.said != "" { + doneKeysSay("done keys: " + r.said) + } + // Cleared on every take-in, whoever raised it: a controller before a restart, whose key was delivered again. + if keeper := conditionsFrom; keeper != nil { + if _, err := keeper.Clear(ctx, o.Key(), "composed and granted"); err == nil { + delete(t.raised, r.module) + } + } + case r.nothingNew: + nothing++ + } + } + for _, m := range batch { + if verdicts[m.module] == refused { + m.nak(doneKeyRetry) + } else { + m.ack() + } + } + return fmt.Sprintf("%d key(s) of %d module(s): %d taken in, %d with nothing new, %d waiting, %d failed; %d composer pass(es)", + len(batch), len(settled), taken, nothing, waiting, failed, passes) +} + +// takeInBacklog is the take-in's backlog on the bus: keys not delivered yet, keys delivered and not acknowledged, and +// the last delivered sequence of each consumer, which moves when anything is delivered. +type takeInBacklog struct { + Pending uint64 `json:"pending"` + Taking uint64 `json:"taking"` + delivered []uint64 +} + +// Total is every key the take-in has not acknowledged. +func (b takeInBacklog) Total() uint64 { return b.Pending + b.Taking } + +// backlogOf reads the consumers' own counts from the bus. +func backlogOf(ctx context.Context, consumers []jetstream.Consumer) (takeInBacklog, error) { + var b takeInBacklog + for _, c := range consumers { + info, err := c.Info(ctx) + if err != nil { + return takeInBacklog{}, err + } + b.Pending += info.NumPending + b.Taking += uint64(info.NumAckPending) + b.delivered = append(b.delivered, info.Delivered.Consumer) + } + return b, nil +} + +// readTakeInBacklog is the take-in's backlog for a reader in any process (`status`): the consumers asked by name. +func readTakeInBacklog(ctx context.Context) (takeInBacklog, error) { + keys, closeKeys, err := openBuilderKeys() + if err != nil { + return takeInBacklog{}, err + } + defer closeKeys() + var consumers []jetstream.Consumer + for _, bucket := range broker.TakeInBuckets { + c, err := keys.JS.Consumer(ctx, "KV_"+bucket, broker.TakeInConsumer) + if err != nil { + return takeInBacklog{}, fmt.Errorf("the take-in's consumer on %s: %w", bucket, err) + } + consumers = append(consumers, c) + } + return backlogOf(ctx, consumers) +} + +// kindTakeInStalled is the condition of a take-in whose backlog stood still (ADR 0544 §6). +const kindTakeInStalled = "take-in-stalled" + +func takeInStalledObservation(b takeInBacklog, since time.Time) conditions.Observation { + return conditions.Observation{Scope: conditions.ScopeCore, ID: "take-in", Token: kindTakeInStalled, + Kind: kindTakeInStalled, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, Source: "done keys", + Summary: fmt.Sprintf("the take-in of done keys has %d key(s) waiting and has taken none in since %s", + b.Total(), since.UTC().Format(time.RFC3339))} +} + +// stallWatch judges the backlog read at each look: stalled once it stood above zero, with nothing delivered, for +// takeInStall. +type stallWatch struct { + mu sync.Mutex + last []uint64 + movedAt time.Time + stalled bool + lastTotal uint64 +} + +// look takes a reading at now and answers whether the take-in stalls and whether that changed. +func (w *stallWatch) look(b takeInBacklog, now time.Time) (stalled, changed bool) { + w.mu.Lock() + defer w.mu.Unlock() + moved := b.Total() == 0 || len(w.last) != len(b.delivered) + for i := range b.delivered { + if !moved && b.delivered[i] != w.last[i] { + moved = true + } + } + if moved || w.movedAt.IsZero() { + w.movedAt = now + } + w.last, w.lastTotal = b.delivered, b.Total() + nowStalled := b.Total() > 0 && now.Sub(w.movedAt) >= takeInStall + changed = nowStalled != w.stalled + w.stalled = nowStalled + return nowStalled, changed +} + +// watchTakeInStall reads the backlog every minute and raises take-in-stalled while it stands still, clearing it when it +// moves: a watchdog's timer, as research 072 §2 keeps them until the health part leaves the controller. +func watchTakeInStall(ctx context.Context, backlog func(context.Context) (takeInBacklog, error)) { + w := &stallWatch{} + tick := time.NewTicker(time.Minute) + defer tick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-tick.C: + } + b, err := backlog(ctx) + if err != nil { + continue + } + stalled, changed := w.look(b, time.Now()) + keeper := conditionsFrom + if !changed || keeper == nil { + continue + } + o := takeInStalledObservation(b, w.movedAt) + if stalled { + doneKeysSay("done keys: " + o.Summary) + _, _ = keeper.Observe(ctx, o) + } else { + _, _ = keeper.Clear(ctx, o.Key(), "the take-in moves again") + } + } +} diff --git a/internal/broker/composed.go b/internal/broker/composed.go index 8856186e..b950e058 100644 --- a/internal/broker/composed.go +++ b/internal/broker/composed.go @@ -182,3 +182,11 @@ func RaiseAfterDeclarations(r Raiser) error { } return AssertMeshConsumers(r) } + +// TakeInConsumer is the durable consumer the controller takes done keys in from, on the stream of each of TakeInBuckets +// (novox/hq ADR 0544 §1). +const TakeInConsumer = "mesh-controller-take-in" + +// TakeInBuckets are the buckets the take-in consumes: the indexer's modules (each module's trunk value) and the builder's +// builds (each build's key). +var TakeInBuckets = []string{delivery.ModulesBucket, delivery.BuilderBuildsBucket} diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 2add5851..734bffc1 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -432,6 +432,16 @@ var DegradableGrants = []DegradableGrant{ Reason: "carried only at genesis: the controller composes its own permissions into the bus module's composition, " + "and writes a composition's digest only after the bus applied it (novox/hq ADR 0539 clauses 2 and 7)", }, + { + Subject: "$JS.ACK.KV_" + delivery.ModulesBucket + "." + TakeInConsumer + ".>", + Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " + + "batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)", + }, + { + Subject: "$JS.ACK.KV_" + delivery.BuilderBuildsBucket + "." + TakeInConsumer + ".>", + Reason: "refused: an acknowledgement refused leaves the key delivered again after its wait, and the take-in's first " + + "batch with something new composes the bus's node, which grants it (novox/hq ADR 0544 §1)", + }, { Subject: "$KV." + GrantsBucket + ".>", Reason: "carried only at genesis: the controller composes its own permissions and pushes them to the bus at its " + @@ -707,6 +717,11 @@ func PermissionsFor(p Principal) (Permissions, error) { // ever, refused by the list it already has. sub = append(sub, ControllerFollows...) pub = append(pub, "$JS.ACK.EVENTS."+ControllerName+".>", "$JS.ACK.EVENTS."+MergesConsumerName+".>") + // And the take-in's two durable consumers on the indexer's and the builder's buckets (novox/hq ADR 0544 §1): + // acknowledged after each key is taken in, and no other consumer of those streams. + for _, bucket := range TakeInBuckets { + pub = append(pub, "$JS.ACK.KV_"+bucket+"."+TakeInConsumer+".>") + } // **Where an enrolment's answer goes**, and `allow_responses` does not cover it. That // permits one reply to the reply subject of a message the user received — and a message a diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 0cc3eab2..7560fd8c 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -29,7 +29,7 @@ accounts { ] users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.ACK.KV_mesh-delivery-builder_builds.mesh-controller-take-in.>", "$JS.ACK.KV_mesh-delivery-indexer_modules.mesh-controller-take-in.>", "$JS.API.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_builds.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_composed.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_declared-packages.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_grants.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollouts.>", "$KV.mesh-controller_states-announced.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.mod.mesh-controller.event.states", "mesh.node.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.artifacts-gone", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.bucket-deleted", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-controller.tool.push", "mesh.seat.mesh-controller.tool.upgrade", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } }