diff --git a/cmd/mesh-controller/composed_keys.go b/cmd/mesh-controller/composed_keys.go index 58348ae1..d83ec114 100644 --- a/cmd/mesh-controller/composed_keys.go +++ b/cmd/mesh-controller/composed_keys.go @@ -40,8 +40,11 @@ type takeInSteps struct { 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) + // registered says whether the module is registered at the trunk's commit, and whether at its fingerprint: a commit + // that moves while the fingerprint stays (a catalogue merge that leaves the module's sources alone) is no new build. + registered func(ctx context.Context, module string, trunk sdkdelivery.Module) (atCommit, atFP bool, err error) + // recordCommit records the module's registration at the trunk's commit, its fingerprint unchanged: nothing composed. + recordCommit func(ctx context.Context, module, commit string) error // 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 // taken says the module's compositions at fp exist for every node and its grant is written: nothing to compose. @@ -151,7 +154,14 @@ func (s takeInSteps) batch(ctx context.Context, keys []delivered, memo map[strin out = append(out, r) continue } - at, err := s.registeredAt(ctx, m, trunk.Commit) + at, atFP, err := s.registered(ctx, m, trunk) + if err == nil && !at && atFP { + // Only the commit moved (issue 1945: every catalogue merge took about 200 modules in again, 25 minutes): the + // new commit is recorded, and the module is judged as one registered, composed only if something is missing. + if err = s.recordCommit(ctx, m, trunk.Commit); err == nil { + at = true + } + } if err != nil { r.verdict, r.err = refused, err out = append(out, r) @@ -311,12 +321,19 @@ func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) ta trunk: func(ctx context.Context, module string) (sdkdelivery.Module, error) { return keys.ModuleAt(ctx, module, "") }, - registeredAt: func(ctx context.Context, module, commit string) (bool, error) { + registered: func(ctx context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) { src, err := inv.SourceOf(ctx, module) if err != nil { - return false, err + return false, false, err } - return sameCommitID(src.BuiltFrom, commit), nil + if sameCommitID(src.BuiltFrom, trunk.Commit) { + return true, true, nil + } + fp, err := registeredFingerprint(ctx, inv, keys, module, src.BuiltFrom) + return false, err == nil && fp != "" && fp == trunk.FP2, nil + }, + recordCommit: func(ctx context.Context, module, commit string) error { + return inv.RecordBuiltFrom(ctx, module, commit) }, takeIn: func(ctx context.Context, module string, trunk sdkdelivery.Module) error { _, _, err := takeInKey(ctx, inv, keys, keyTake{Module: module, Trunk: trunk, Ask: registrationAsk{Merge: trunk.Commit}}) @@ -343,6 +360,24 @@ func liveTakeInSteps(open *stores, keys registerKeys, js jetstream.JetStream) ta } } +// registeredFingerprint is the fingerprint a module is registered at, as currentFingerprints reads it: its commit's +// newest build's, or the indexer's value at the commit when the commit has no build of its own; "" when neither says. +func registeredFingerprint(ctx context.Context, inv *inventory.Inventory, keys registerKeys, module, commit string) (string, error) { + if commit == "" { + return "", nil + } + if id, err := inv.BuildOf(ctx, module, commit); err == nil && id != "" { + if b, found, err := inv.BuildByID(ctx, id); err == nil && found && b.SourceFingerprint != "" { + return b.SourceFingerprint, nil + } + } + at, err := keys.ModuleAt(ctx, module, commit) + if err != nil { + return "", err + } + return at.FP2, nil +} + // 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) { diff --git a/cmd/mesh-controller/composed_keys_test.go b/cmd/mesh-controller/composed_keys_test.go index ddd86bd1..9e133ce5 100644 --- a/cmd/mesh-controller/composed_keys_test.go +++ b/cmd/mesh-controller/composed_keys_test.go @@ -19,6 +19,7 @@ import ( // 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 + sameFP map[string]bool // registered at another commit, at the trunk's fingerprint trunks map[string]string // module -> commit; no entry: no trunk value nodes map[string][]string taken map[string]bool @@ -47,8 +48,15 @@ func (m *aTakeIn) steps() takeInSteps { } return sdkdelivery.Module{Commit: commit, FP2: fp(module)}, nil }, - registeredAt: func(_ context.Context, module, commit string) (bool, error) { - return m.registered[module] == commit, nil + registered: func(_ context.Context, module string, trunk sdkdelivery.Module) (bool, bool, error) { + if m.registered[module] == trunk.Commit { + return true, true, nil + } + return false, m.sameFP[module], nil + }, + recordCommit: func(_ context.Context, module, commit string) error { + m.acts = append(m.acts, "record "+module+" at "+commit) + return nil }, takeIn: func(_ context.Context, module string, trunk sdkdelivery.Module) error { m.acts = append(m.acts, "take in "+module+" at "+trunk.Commit) @@ -113,6 +121,12 @@ func TestATakeInBatch(t *testing.T) { keys: []delivered{{module: "mail"}, {module: "nats"}, {module: "mail"}}, acts: []string{"catalogue"}, want: map[string]want{"mail": {nothingNew: true}, "nats": {nothingNew: true}}}, + {name: "a catalogue merge that moved the commit and not the fingerprint: recorded, nothing composed", + mesh: aTakeIn{registered: map[string]string{"mail": "a85d9037"}, sameFP: map[string]bool{"mail": true}, + trunks: map[string]string{"mail": "2df05030"}, nodes: map[string][]string{"mail": {"ace"}}, + taken: map[string]bool{"mail": true}, holder: "novox"}, + keys: []delivered{{module: "mail"}}, + acts: []string{"catalogue", "record mail at 2df05030"}, want: map[string]want{"mail": {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"}, @@ -270,6 +284,10 @@ func TestATakeInStallsWhenItHoldsWorkAndSettlesNone(t *testing.T) { {0, 0, 3, 40, 40, false, true}, {30 * time.Minute, 0, 3, 40, 40, false, false}, }}, + {"every key settled, one of them refused by the bus, nothing pending: no stall", []reading{ + {0, 0, 1, 3, 3, false, true}, + {6 * time.Minute, 0, 1, 3, 3, false, false}, + }}, {"a batch handed and never settled is a stall", []reading{ {0, 0, 16, 56, 40, false, true}, {5 * time.Minute, 0, 16, 56, 40, true, true}, diff --git a/cmd/mesh-controller/take_in_consumer.go b/cmd/mesh-controller/take_in_consumer.go index dd6023cc..8ae7f4b7 100644 --- a/cmd/mesh-controller/take_in_consumer.go +++ b/cmd/mesh-controller/take_in_consumer.go @@ -133,6 +133,9 @@ func watchDoneKeys(ctx context.Context, open *stores) { // it): what tells a take-in that moves from one that holds work and does nothing (the stall watch, ADR 0544 §6). moving := &takeInMoving{} read := make([]jetstream.Consumer, 0, len(consumers)) + // The connection closes only once the readers and the loop have ended, so the refusals they send at the stop reach + // the bus (the third review of ADR 0544). + var users sync.WaitGroup for _, c := range consumers { bucket := c.bucket read = append(read, c.consumer) @@ -145,7 +148,9 @@ func watchDoneKeys(ctx context.Context, open *stores) { <-ctx.Done() it.Drain() // the keys it holds are handed out and refused at once below, not lost to their ack wait }) + users.Add(1) goServing(ctx, func() { + defer users.Done() for { m, err := it.Next() if err != nil { @@ -159,16 +164,20 @@ func watchDoneKeys(ctx context.Context, open *stores) { _ = m.Ack() continue } + // Settled is the loop's attempt, whatever the bus answers: an acknowledgement it refuses (at a first start, + // until the bus node is composed) is delivered again and settled again, and never counts as work held + // (the third review of ADR 0544). Acknowledged synchronously and past the stop, so a refusal is said. msg := takeInMsg{delivered: d, ack: func() { - // Counted only when the bus took it: at a first start it refuses the acks until the bus node is composed. - if m.DoubleAck(ctx) == nil { - moving.settled.Add(1) + acking, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + if err := m.DoubleAck(acking); err != nil { + doneKeysSay(fmt.Sprintf("done keys: the bus did not take the acknowledgement of %s: %v", d.module, err)) } + moving.settled.Add(1) }, nak: func(delay time.Duration) { - if m.NakWithDelay(delay) == nil { - moving.settled.Add(1) - } + _ = m.NakWithDelay(delay) + moving.settled.Add(1) }} if ctx.Err() != nil { _ = m.Nak() @@ -186,10 +195,15 @@ func watchDoneKeys(ctx context.Context, open *stores) { } 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) }) + users.Add(1) + goServing(ctx, func() { + defer users.Done() + takeInLoop(ctx, in, steps, backlog) + }) goServing(ctx, func() { watchTakeInStall(ctx, backlog, moving) }) goServing(ctx, func() { <-ctx.Done() + users.Wait() closeKeys() }) } diff --git a/internal/inventory/catalogue.go b/internal/inventory/catalogue.go index c4aa198a..cebcd655 100644 --- a/internal/inventory/catalogue.go +++ b/internal/inventory/catalogue.go @@ -210,6 +210,21 @@ func (i *Inventory) hasModule(ctx context.Context, name string) (bool, error) { // This is the whole of noticing. Nothing here builds anything — it writes down that the two // halves differ, which is what makes *is this current?* answerable without building, and what // makes a module that nobody rebuilt visible rather than silent. +// RecordBuiltFrom records that a registered module stands at commit, its manifest and fingerprint unchanged: a catalogue +// merge that left its sources alone (novox/hq ADR 0544 §3). The head moves with it, as a registration moves it; the +// module's builds and manifest stay as they are. +func (i *Inventory) RecordBuiltFrom(ctx context.Context, module, commit string) error { + tag, err := i.store.Pool().Exec(ctx, + `update module set built_from = $2, source_head = $2 where name = $1`, module, commit) + if err != nil { + return fmt.Errorf("cannot record %s at %s: %w", module, commit, err) + } + if tag.RowsAffected() == 0 { + return fmt.Errorf("%w: %s", ErrNoSuchModule, module) + } + return nil +} + func (i *Inventory) SourceMoved(ctx context.Context, module, head string) error { tag, err := i.store.Pool().Exec(ctx, `update module set source_head = $2, source_seen = now() where name = $1`, module, head)