diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index 2c98c41a..6bce7560 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -216,7 +216,7 @@ var errReadOnly = errors.New("the tools bundle only reads this state") func main() { ask := askTakeIn(stdio.State(index.TakeInsState), 25*time.Second) tools := append(Tools(ask, time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...) - tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, ask, time.Now)...) + tools = append(tools, index.PendingTools(keptRead{stdio.State(index.PendingState)}, keptRead{stdio.State(index.OwnBuildStateName)}, ask, time.Now)...) tools = append(tools, index.TryCandidateTools(keptRead{stdio.State("candidates")}, keptRead{stdio.State(index.TryRecoveryState)}, ask, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index 023478a4..e7feb4a1 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -283,7 +283,7 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10; 1585/T14, T15 } waits := func() []takein.PendingWait { t.Helper() - answer, err := index.PendingTools(kvStore{pendingKV}, nil, time.Now)[0].Run(map[string]any{}) + answer, err := index.PendingTools(kvStore{pendingKV}, nil, nil, time.Now)[0].Run(map[string]any{}) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 9bf29c33..5445dc48 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -343,8 +343,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro oldSections := process.AwaitBucket(ctx, js, conn, index.OldSectionsBucket, "the count of the old sections is not kept, so the judge refuses no use of one", logger.Printf) // The module graph (ADR 0603 §2): a state this delivery brings, bound when it appears; until then no graph is written. graph := process.AwaitBucket(ctx, js, conn, graph.Bucket, "no module graph is written", logger.Printf) + ownBuildState := process.AwaitBucket(ctx, js, conn, index.OwnBuildBucket, "which build this indexer runs is not kept", logger.Printf) tryRecovery := process.AwaitBucket(ctx, js, conn, index.TryRecoveryBucket, "how the recovery of the open tries went is not kept", logger.Printf) - ix := &index.Indexer{TryRecovery: awaitedStore{tryRecovery}, OldSections: awaitedStore{oldSections}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, + ix := &index.Indexer{OwnBuildState: awaitedStore{ownBuildState}, TryRecovery: awaitedStore{tryRecovery}, OldSections: awaitedStore{oldSections}, Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: awaitedStore{taken}, Builds: known, Verdicts: awaitedStore{verdicts}, SayRefused: sayRefused(js, cred, logger.Printf), SayAccepted: sayAccepted(js, cred, logger.Printf), Predefined: awaitedStore{predefined}, ConfigCandidates: awaitedStore{configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), @@ -354,6 +355,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // until then every merge is taken in at once, as before. Pending: awaitedStore{process.AwaitBucket(ctx, js, conn, index.PendingBucket, "no merge is held for the indexer its revision names: each is taken in at once", logger.Printf)}, SayTakeIn: sayTakeIn(js, cred)} + // Which build this program is, fixed here once, before it hears any merge and never asked again (index.FixOwnBuild): the + // node-engine names the successor while it replaces this process, and an old process that learnt it later would take + // revisions in with its old judge. + ix.FixOwnBuild(ctx) // The log's entries at the start are read before any fingerprint is computed or registered module indexed (below); each // new one indexes the module it names when that module is registered and the bucket holds none (registered.go). var started atomic.Bool diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index bbae2758..137609cf 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -122,12 +122,17 @@ type Indexer struct { fillAgain atomic.Bool // a kick arrived while it ran atHeads map[string]map[string]delivery.PublishedPackages // a member head's published Go modules, by repository@head (candidatePublished) // OwnBuild answers the fingerprint of the build this program runs, from the node-engine's seat; nil knows none. - OwnBuild func(ctx context.Context) (string, error) - underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go) - judgeMu sync.Mutex // one judge pass at a time (verdicts.go) - ownMu sync.Mutex // guards ownFP - ownFP string // the fingerprint of the build this program runs, asked at the start pass (learnOwnBuild) - judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go) + OwnBuild func(ctx context.Context) (string, error) + underSeats bool // a dry run inside judge.WithSeats: its judge passes do not hold the seat list (verdicts.go) + judgeMu sync.Mutex // one judge pass at a time (verdicts.go) + ownMu sync.Mutex // guards ownFP + ownFP string // the fingerprint of the build this program runs, fixed once at the start (FixOwnBuild) + ownFixed bool // FixOwnBuild asked already: it asks once + ownSince time.Time // when the build was asked and not known, with ownWhy + ownWhy string + // OwnBuildState keeps which build this process knows it runs, or since when and why not (FixOwnBuild); nil keeps none. + OwnBuildState Store + judgePending atomic.Bool // the start's judge pass failed: judged again at the next message (verdicts.go) // Message asks the forge bridge for the merged message of a merge, repository and merge commit in, the message whole out (a // pending value keeps a reference, not a copy: issue 1585); an error is the bridge's words. Nil asks none. Message func(ctx context.Context, repository, merge string) ([]byte, error) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index 39189ff2..cd34f376 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -339,20 +339,27 @@ func (ix *Indexer) pendingOnBus(ctx context.Context) bool { return !errors.Is(err, process.ErrNotYet) } -// ownBuildNow is the build this program runs: the one learnt, else asked of the node-engine now. -func (ix *Indexer) ownBuildNow(ctx context.Context) string { - ix.ownMu.Lock() - fp := ix.ownFP - ix.ownMu.Unlock() - if fp != "" || ix.OwnBuild == nil { - return fp - } - ix.learnOwnBuild(ctx) +// ownBuildNow is the build this program runs, as FixOwnBuild fixed it at the start: empty while unknown, and never asked again. +func (ix *Indexer) ownBuildNow(context.Context) string { ix.ownMu.Lock() defer ix.ownMu.Unlock() return ix.ownFP } +// sayOwnBuildUnknown says, for a merge this process cannot take in because it does not know its own build (ADR 0633 §1), a +// take-in.pending that names why and since when, so the watchman's take-in-waits bound covers it. The merge is not written to +// pending; the stream delivers it again to the process that knows its build. +func (ix *Indexer) sayOwnBuildUnknown(ctx context.Context, seq uint64, f forgemsg.Forge) { + ix.ownMu.Lock() + since, why := ix.ownSince, ix.ownWhy + ix.ownMu.Unlock() + if since.IsZero() { // never fixed: the start did not run (no engine wired) + since, why = ix.now(), "the build this program runs was never asked" + } + ix.sayTakeIn(ctx, takein.TakeIn{Kind: takein.TakeInPending, Key: takein.PendingKey(seq), What: takein.TakeInRevision, + Merges: []takein.Merge{{Repository: f.Repository, Number: f.Number, Merge: f.Merge}}, Since: since, Unknown: why}) +} + // sayTakeIn publishes a take-in event and keeps what it said on the entries, so indexer_pending reads the last event of // each wait from the state. A publish that fails is said in the log: the next evaluation says it again. func (ix *Indexer) sayTakeIn(ctx context.Context, t takein.TakeIn) bool { @@ -401,6 +408,7 @@ func (ix *Indexer) Arrive(ctx context.Context, seq uint64, f forgemsg.Forge) (Ap } own := ix.ownBuildNow(ctx) if own == "" { + ix.sayOwnBuildUnknown(ctx, seq, f) return Applied{}, "", ErrOwnBuildUnknown } ix.pendingMu.Lock() @@ -691,7 +699,7 @@ func (ix *Indexer) dry() *dryRun { d := &dryRun{} x := &Indexer{Manifests: ix.Manifests, Packages: ix.Packages, JudgeID: ix.JudgeID, Files: ix.Files, GoSum: ix.GoSum, GoMod: ix.GoMod, Now: ix.Now, Builds: ix.Builds, Named: ix.Named, Trunk: ix.Trunk, Read: ix.Read, Contains: ix.Contains, - ReadOpen: nil, OwnBuild: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil, ReadFile: ix.ReadFile, underSeats: true} + ReadOpen: nil, OwnBuild: nil, OwnBuildState: nil, Message: ix.Message, SayRefused: nil, SayAccepted: nil, SayTakeIn: nil, Pending: nil, ReadFile: ix.ReadFile, underSeats: true} x.Logf = func(format string, args ...any) { ix.logf("take-in, computed and not written: "+format, args...) } x.Store, d.candidates = over(ix.Store) x.Modules, d.modules = over(ix.Modules) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index b7c1ba71..7f787eb7 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -61,6 +61,8 @@ func newTakeInBed(t *testing.T) *takeInBed { t.Fatalf("the bed's modules have no fingerprint") } ix.OwnBuild = func(context.Context) (string, error) { return own, nil } + ix.OwnBuildState = newMem() + ix.FixOwnBuild(context.Background()) // fixed once, at the start return b } @@ -131,11 +133,26 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) ctx := context.Background() t.Run("T6 a merge while the own build is unknown writes nothing", func(t *testing.T) { b := newTakeInBed(t) + b.ix.ownFP, b.ix.ownFixed = "", false // this process is started again and asks the engine, which does not answer b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") } + b.ix.FixOwnBuild(ctx) + b.said = nil _, _, err := b.ix.Arrive(ctx, 1, b.merge(20, sha('b'), "tree-avahi-2", "")) if !errors.Is(err, ErrOwnBuildUnknown) || len(b.keys()) != 0 || b.commit(t, "avahi") == sha('b') { t.Errorf("err %v, pending %v, avahi at %s: want ErrOwnBuildUnknown and nothing written", err, b.keys(), b.commit(t, "avahi")) } + // The merge is said pending with why and since when (so the watchman's bound covers it), and indexer_pending says the + // build is unknown, from the own-build state. + if len(b.said) != 1 || b.said[0].Kind != takein.TakeInPending || b.said[0].Key != takein.PendingKey(1) || !strings.Contains(b.said[0].Unknown, "does not answer") || + b.said[0].Since.IsZero() || b.said[0].WaitsFor != "" { + t.Errorf("said %+v, want one take-in.pending of m.1 saying why the own build is unknown", b.said) + } else if _, err := takein.EncodeTakeIn(b.said[0]); err != nil { + t.Errorf("the unknown-build event is refused by its own contract: %v", err) + } + answer, err := PendingTools(b.pending, b.ix.OwnBuildState, nil, nil)[0].Run(map[string]any{}) + if a, _ := answer.(takein.PendingAnswer); err != nil || !strings.Contains(a.OwnBuildUnknown, "does not answer") || a.OwnBuildSince.IsZero() { + t.Errorf("indexer_pending = %+v %v, want the build unknown with why and since", answer, err) + } // Before the pending state is on the bus nothing waits: the start asks no build, and the merge is taken in at once, // as before ADR 0633, though the build is unknown (TestIndexerWhole found the start waiting on it for ever). b.ix.Pending = &memStore{unbound: process.ErrNotYet} @@ -254,31 +271,36 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) g.Members = []delivery.Head{{Repository: "novox/mesh-sdk", Number: 30, Head: g.Head}, {Repository: "novox/mesh-host", Number: 4, Head: sha('9')}} b.arrive(t, 5, g) named := b.fp(t, delivery.IndexerModule) - // The new indexer starts; its build is not known at first: a hand ask is answered meanwhile (R5), and the start waits. - next := &Indexer{} - *next = Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts, - Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum} - next.toolchains = b.ix.toolchains - asked := 0 - next.OwnBuild = func(context.Context) (string, error) { - if asked++; asked < 3 { // the hand ask's and the first start's - return "", errors.New("not yet") - } - return named, nil + // The new indexer starts; its build is not known (the engine does not answer at its start): a hand ask is answered + // meanwhile (R5), and the start waits for nothing. It is never asked again, so it stays unknown until it is started again. + mk := func(ask func(context.Context) (string, error)) *Indexer { + n := &Indexer{Store: b.ix.Store, Modules: b.ix.Modules, Manifests: b.ix.Manifests, Graph: b.ix.Graph, Verdicts: b.ix.Verdicts, + Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: b.ix.Message, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum} + n.toolchains = b.ix.toolchains + n.OwnBuild = ask + n.FixOwnBuild(ctx) + return n } + asked := 0 + next := mk(func(context.Context) (string, error) { asked++; return "", errors.New("not yet") }) if r := next.TakeInPendingByHand(ctx, "m.nothing", "lead@shanks", "a test"); r.Stored || !strings.Contains(r.Answer, "nothing waits") { t.Errorf("a hand ask while the own build is unknown: %+v", r) } - // The first start, its build not known, waits for nothing (no sleep, no timer): it answers at once, takes nothing - // in, and the forge consumer starts after it. - if _, words, err := next.TakeInAtStart(ctx); err != nil || asked != 2 || len(b.keys()) != 2 || !strings.Contains(words, "not known") { - t.Fatalf("a start with its build unknown: %q %v, asked %d, pending %v: want it to answer at once and take nothing in", words, err, asked, b.keys()) + if _, words, err := next.TakeInAtStart(ctx); err != nil || len(b.keys()) != 2 || !strings.Contains(words, "not known") { + t.Fatalf("a start with its build unknown: %q %v, pending %v: want it to answer at once and take nothing in", words, err, b.keys()) } + // The engine answers now, and the same process does not ask it again: it learnt nothing it did not know at its start. + next.OwnBuild = func(context.Context) (string, error) { asked++; return named, nil } + if _, words, err := next.TakeInAtStart(ctx); err != nil || asked != 1 || len(b.keys()) != 2 || !strings.Contains(words, "not known") { + t.Fatalf("the window of a replacement: %q %v, asked %d, pending %v: want the build fixed at the start, never learnt later", words, err, asked, b.keys()) + } + // The module started again fixes the build at its start and takes the revision in. + next = mk(func(context.Context) (string, error) { return named, nil }) if _, _, err := next.TakeInAtStart(ctx); err != nil { t.Fatal(err) } - if asked < 3 || !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') { - t.Errorf("start: asked %d, pending %v, avahi at %s: want the revision taken in, the group left", asked, b.keys(), b.commit(t, "avahi")) + if !reflect.DeepEqual(b.keys(), []string{takein.PendingKey(5)}) || b.commit(t, "avahi") != sha('c') { + t.Errorf("start: pending %v, avahi at %s: want the revision taken in, the group left", b.keys(), b.commit(t, "avahi")) } }) t.Run("T8 T9 T11 a group gathered until its end, a merge behind it, one alone, an end again", func(t *testing.T) { @@ -452,7 +474,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2")) b.arrive(t, 3, b.merge(22, sha('d'), "tree-avahi-4", "tree-indexer-2")) var answer takein.PendingAnswer - for _, tool := range PendingTools(b.pending, nil, nil) { + for _, tool := range PendingTools(b.pending, nil, nil, nil) { if tool.Name == PendingTool { out, err := tool.Run(map[string]any{}) if err != nil { @@ -522,6 +544,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) Packages: b.ix.Packages, Builds: b.ix.Builds, Pending: b.pending, Message: bridgeDown, SayTakeIn: b.ix.SayTakeIn, Logf: b.ix.Logf, GoSum: b.ix.GoSum, OwnBuild: func(context.Context) (string, error) { return named, nil }} next.toolchains = b.ix.toolchains + next.FixOwnBuild(ctx) b.said = nil if _, _, err := next.TakeInAtStart(ctx); err != nil { t.Fatal(err) @@ -578,7 +601,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) named := map[string]bool{} for _, f := range []string{"Store", "Modules", "Manifests", "Predefined", "ConfigCandidates", "Taken", "Verdicts", "OldSections", "JudgeID", "SayRefused", "SayAccepted", "Packages", "Graph", "Files", "GoSum", "GoMod", "Now", "Builds", "Named", "Trunk", - "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile", "OpenTries", "TryRecovery"} { + "Read", "Contains", "Logf", "ReadOpen", "OwnBuild", "Message", "Pending", "SayTakeIn", "ReadFile", "OpenTries", "TryRecovery", "OwnBuildState"} { named[f] = true } ty := reflect.TypeOf(Indexer{}) diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index e8811e78..733cb70a 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -16,7 +16,7 @@ import ( // PendingTools are the indexer's tools of what waits for a take-in (novox/hq ADR 0633 §5): indexer_pending reads the // pending state, and indexer_take_in_pending and indexer_drop_pending steer it by an ask in takeins, as indexer_take_in // does. pending is the pending state as the tools bundle reads it; ask writes an ask and waits for the indexer's answer. -func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { +func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (TakeInAsk, error), now func() time.Time) []stdio.Tool { str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s } hand := func(drop bool) func(map[string]any) (any, error) { return func(args map[string]any) (any, error) { @@ -53,7 +53,8 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, "merges, since when it waits, the modules it holds back (holds), the merges whose message the forge bridge's stream no " + "longer holds (lost: they are not taken in; indexer_drop_pending drops them) and the last take-in event said of it. Empty waits: " + "every merge is taken in. The watchman raises take-in-waits past 60 minutes for a revision, 30 for a group; " + - "indexer_take_in_pending and indexer_drop_pending steer a wait.", + "indexer_take_in_pending and indexer_drop_pending steer a wait. own-build-unknown, when set, says the indexer cannot learn the build it " + + "runs (since when): it fixes it once at its start, so every merge waits (a take-in.pending that says so) until the module is started again.", Input: map[string]any{"type": "object", "properties": map[string]any{}}, Run: func(map[string]any) (any, error) { var unread []string @@ -63,7 +64,17 @@ func PendingTools(pending Store, ask func(key string, value []byte) (TakeInAsk, if err != nil { return nil, err } - return takein.PendingAnswer{Waits: PendingWaits(entries), Unread: unread}, nil + answer := takein.PendingAnswer{Waits: PendingWaits(entries), Unread: unread} + // Whether this indexer knows the build it runs: when it does not, every merge waits, and this says since when and why. + if ownBuild != nil { + if v, _, found, err := ownBuild.Get(context.Background(), OwnBuildKey); err == nil && found { + var o OwnBuild + if json.Unmarshal(v, &o) == nil && !o.Known { + answer.OwnBuildUnknown, answer.OwnBuildSince = o.Why, o.Since + } + } + } + return answer, nil }}, {Name: TakeInPendingTool, Description: "Take a wait in by hand with the running indexer and its judge (ADR 0633 §5): the revision under its m. key, " + diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts.go b/modules/mesh-delivery-indexer/internal/index/verdicts.go index 94f65448..1758f740 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts.go @@ -211,7 +211,6 @@ func (ix *Indexer) judgeOverHeld(ctx context.Context, fresh map[string]bool, ove // digest of the verdicts state that another judge refused or held waiting (one of a candidate, or one back on the trunk // later) is judged again by this judge (judgeStale), so a refusal an older judge made never stands for good. func (ix *Indexer) JudgeAtStart(ctx context.Context) (Judged, error) { - ix.learnOwnBuild(ctx) j, err := ix.JudgeAll(ctx, nil) ix.judgePending.Store(err != nil) if err == nil { @@ -355,21 +354,66 @@ func (ix *Indexer) writeVerdict(ctx context.Context, digest string, v delivery.M return wrote, err } -// learnOwnBuild asks where the indexer's program is told which build it runs (OwnBuild: the node-engine's seat, never the -// trunk's value of the indexer, which names a build that may be merged and not delivered). Asked at the start pass; where -// it is not answered the build stays unknown and every unknown field waits, which is the safe side. -func (ix *Indexer) learnOwnBuild(ctx context.Context) { - if ix.OwnBuild == nil { +// OwnBuildState is the indexer's state of the build it runs (the bucket mesh-delivery-indexer_own-build, one key, OwnBuildKey): +// written once, at the process's start by FixOwnBuild, so indexer_pending says whether this process knows its build and, when it +// does not, since when and why. +const ( + OwnBuildStateName = "own-build" + OwnBuildBucket = delivery.IndexerModule + "_" + OwnBuildStateName + OwnBuildKey = "state" +) + +// OwnBuild is the value of the own-build state. +type OwnBuild struct { + V int `json:"v"` + Known bool `json:"known"` + FP string `json:"fp,omitempty"` + Since time.Time `json:"since"` + Why string `json:"why,omitempty"` +} + +// FixOwnBuild fixes, once and at the process's start, which build of the indexer this program is: asked of the node-engine's +// seat (OwnBuild; never the trunk's value of the indexer, which names a build that may be merged and not delivered), before +// this process hears a merge. It is never asked again: while the node-engine replaces the indexer the seat already names the +// successor, and an old process that learnt it later would take revisions in with its old judge. Where the start's ask is not +// answered the build stays unknown for the life of this process: every merge waits, loudly (a take-in.pending that says so, the +// own-build state, indexer_pending), and the module is started again to learn it. +func (ix *Indexer) FixOwnBuild(ctx context.Context) { + ix.ownMu.Lock() + if ix.ownFixed || ix.OwnBuild == nil { + ix.ownMu.Unlock() return } + ix.ownFixed = true + ix.ownMu.Unlock() fp, err := ix.OwnBuild(ctx) - if err != nil || fp == "" { - ix.logf("judge: the build this program runs is not known (%v), so a manifest with a field it does not know waits and is not refused", err) - return + if err == nil && fp == "" { + err = errors.New("the node-engine names no build of the indexer") + } + st := OwnBuild{V: 1, Since: ix.now().UTC()} + if err != nil { + st.Why = err.Error() + ix.logf("judge: the build this program runs is not known (%v), so every merge waits and a manifest with a field it does not know waits and is not refused; it is learnt only at the start, so the indexer is started again", err) + } else { + st.Known, st.FP = true, fp } ix.ownMu.Lock() - ix.ownFP = fp + ix.ownFP, ix.ownSince, ix.ownWhy = fp, st.Since, st.Why ix.ownMu.Unlock() + if ix.OwnBuildState == nil { + return + } + data, _ := json.Marshal(st) + if _, rev, found, gerr := ix.OwnBuildState.Get(ctx, OwnBuildKey); gerr == nil && found { + err = ix.OwnBuildState.Update(ctx, OwnBuildKey, data, rev) + } else if gerr == nil { + err = ix.OwnBuildState.Create(ctx, OwnBuildKey, data) + } else { + err = gerr + } + if err != nil { + ix.logf("the own-build state cannot be written: %v", err) + } } // ownBuildIsNewest says whether the build of the indexer this program runs (ownFP, from the node-engine) is the newest the @@ -510,8 +554,6 @@ func (ix *Indexer) JudgeAgain(ctx context.Context, module, digest string) TakenI out.Answer = "the modules state cannot be read: " + err.Error() return out } - // The build this program runs is asked again, so a start pass that missed the engine's answer is recovered without a restart. - ix.learnOwnBuild(ctx) if digest != "" { // One digest (a candidate's, or one the trunk no longer names), judged by this judge in the catalogue the trunk holds. value, _, found, err := ix.Manifests.Get(ctx, digest) diff --git a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go index 6dd718a7..781bc576 100644 --- a/modules/mesh-delivery-indexer/internal/index/verdicts_test.go +++ b/modules/mesh-delivery-indexer/internal/index/verdicts_test.go @@ -273,12 +273,16 @@ func TestTheIndexerJudgesEveryManifestOnce(t *testing.T) { if rows, _ := ListVerdicts(keys, get, "", true); len(rows) <= len(keys)-3 || rows[0].Module > rows[len(rows)-1].Module { t.Fatalf("indexer_verdicts, all: %d rows of %d verdicts", len(rows), len(keys)) } - // indexer_judge asks the engine for the running build again: a start pass that missed it is recovered (the trunk's value of - // the indexer is the newer one, so the field waits; asked and answered with that same build it is refused). - ix.ownFP = "" + // The build is fixed once, at the start, and never learnt later: this program started when the engine could not answer, and + // the engine now names the trunk's newest build (as it does while it replaces the indexer by a successor). indexer_judge + // does not take it for this program's own, so the field still waits and the build stays unknown. + ix.ownFP, ix.ownFixed = "", true ix.OwnBuild = func(context.Context) (string, error) { return "src2:" + strings.Repeat("a", 64), nil } - if got := ix.JudgeAgain(ctx, "later", ""); !got.Stored || !strings.Contains(got.Answer, "refused") || ix.ownFP == "" { - t.Fatalf("indexer_judge did not learn the running build: %+v", got) + if got := ix.JudgeAgain(ctx, "later", ""); !got.Stored || !strings.Contains(got.Answer, "waits") || ix.ownFP != "" { + t.Fatalf("indexer_judge learnt the build the engine names now: %+v (own %q)", got, ix.ownFP) + } + if _, err := ix.JudgeAtStart(ctx); err != nil || ix.ownFP != "" || !verdict(later).Waits() { + t.Fatalf("a judge pass learnt the build: %v %q %+v", err, ix.ownFP, verdict(later)) } ix.ownFP = ownFP ix.OwnBuild = nil diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 731bfa52..3e536b66 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -108,7 +108,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build t.Errorf("emits = %v, want manifest-refused, manifest-accepted and the two take-in events", m.Emits) } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. - if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, + if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: "try-recovery", History: 1}, {Name: "own-build", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, {Name: graph.State, History: graph.History}, {Name: index.PendingState, History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index eabac22c..1791056e 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -47,6 +47,7 @@ { "name": "verdicts", "history": 5 }, { "name": "oldsections", "history": 1 }, { "name": "try-recovery", "history": 1 }, + { "name": "own-build", "history": 1 }, { "name": "packages", "history": 5 }, { "name": "graph", "history": 5 }, { "name": "pending", "history": 1 } diff --git a/modules/mesh-delivery-indexer/takein/takein.go b/modules/mesh-delivery-indexer/takein/takein.go index 3e76e4ba..af9ba44f 100644 --- a/modules/mesh-delivery-indexer/takein/takein.go +++ b/modules/mesh-delivery-indexer/takein/takein.go @@ -140,6 +140,9 @@ type TakeIn struct { // the seat definitions the revision carries (ADR 0633 §1), the manifest validator's problems. The revision waits for // an indexer whose manifest is accepted, and says why here, as it says what it lost. Refused string `json:"refused,omitempty"` + // Unknown is why the indexer holds a merge it does not take in although no indexer build is named: it cannot learn the build + // it runs itself (ADR 0633 §1), so Since is when it began not to know. A pending revision names WaitsFor or Unknown. + Unknown string `json:"own-build-unknown,omitempty"` } func (t TakeIn) check() error { @@ -168,8 +171,8 @@ func (t TakeIn) check() error { if !strings.HasPrefix(t.Key, "m.") { return fmt.Errorf("delivery: a revision's take-in key is its oldest merge's m., not %q", t.Key) } - if t.Kind == TakeInPending && t.WaitsFor == "" { - return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for", t.Key) + if t.Kind == TakeInPending && t.WaitsFor == "" && t.Unknown == "" { + return fmt.Errorf("delivery: a pending revision %s names the indexer build it waits for, or why the indexer does not know its own", t.Key) } case TakeInGroup: if t.Group == "" { @@ -247,6 +250,10 @@ type PendingWait struct { type PendingAnswer struct { Waits []PendingWait `json:"waits"` Unread []string `json:"unread,omitempty"` + // OwnBuildUnknown is why the running indexer does not know its own build, and since when: while it is set, every merge waits + // (a take-in.pending with the same words) and the indexer is started again to learn it. Empty when the build is known. + OwnBuildUnknown string `json:"own-build-unknown,omitempty"` + OwnBuildSince time.Time `json:"own-build-since,omitzero"` } // DecodePendingAnswer reads indexer_pending's answer, refusing a wait that is not a revision or a group, or names no diff --git a/modules/mesh-delivery-indexer/takein/takein_test.go b/modules/mesh-delivery-indexer/takein/takein_test.go index f9a5d6cd..a4d8b53b 100644 --- a/modules/mesh-delivery-indexer/takein/takein_test.go +++ b/modules/mesh-delivery-indexer/takein/takein_test.go @@ -20,6 +20,7 @@ func sampleTakeIns(since time.Time) []TakeIn { Members: []delivery.Head{{Repository: "novox/mesh-delivery", Number: 9, Head: "ffff"}, {Repository: "novox/mesh-sdk", Number: 37, Head: "dddd"}}, Since: since}, {Kind: TakeInTaken, Key: PendingKey(12), What: TakeInRevision, Merges: m, WaitsFor: "src2:abc", Since: since, By: "lead-47@shanks", Why: "the named indexer can never run"}, + {Kind: TakeInPending, Key: PendingKey(13), What: TakeInRevision, Merges: m, Unknown: "the engine does not answer", Since: since}, // the indexer cannot learn its own build } } diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index e03c00ec..4ee4fc29 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -106,6 +106,24 @@ func TestTheWatchmansFindings(t *testing.T) { } expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound-10*time.Minute) }, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}}, + {"ADR 0633 §1 (issue 1921): a merge held because the indexer cannot learn its own build is raised past the revision bound from its since, naming why", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { + w.TakeInCaughtUp() + body, err := takein.EncodeTakeIn(takein.TakeIn{Kind: takein.TakeInPending, Key: revKey, What: takein.TakeInRevision, Since: c.t, + Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}, Unknown: "the engine does not answer"}) + if err != nil { + t.Fatal(err) + } + w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), body) + if o.ttls[signalKey("take-in", revKey)] != TakeInRevisionBound || len(o.findings) != 0 { + t.Fatalf("an unknown own build: ttls %v, findings %v", o.ttls, o.findings) + } + expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound) + for _, f := range o.findings { + if !strings.Contains(f.Summary, "cannot learn the build it runs") { + t.Errorf("finding %q does not say why", f.Summary) + } + } + }, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}}, {"ADR 0633 §5 (1585/T21): a drop clears the wait, and the merges left under it said pending again with the same since are raised from it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) { w.TakeInCaughtUp() // a live message, heard after the replay w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), takeInBody(t, takein.TakeInPending, revKey, t0)) diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 3c3465ff..171aa403 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -72,6 +72,9 @@ func (w *Watchman) takeIn(kind, key string, t takein.TakeIn) { return } bound, what := TakeInRevisionBound, "a source revision waits for the indexer build "+t.WaitsFor + if t.Unknown != "" { + what = "a source revision is held because the indexer cannot learn the build it runs (" + t.Unknown + ")" + } if t.What == takein.TakeInGroup { bound, what = TakeInGroupBound, "a delivery group is gathered and its end has not come" }