diff --git a/cmd/mesh-controller/awaiting_push_integration_test.go b/cmd/mesh-controller/awaiting_push_integration_test.go index ee881d98..d3714080 100644 --- a/cmd/mesh-controller/awaiting_push_integration_test.go +++ b/cmd/mesh-controller/awaiting_push_integration_test.go @@ -10,6 +10,7 @@ import ( "github.com/novox/mesh-controller/internal/catalogue" "github.com/novox/mesh-controller/internal/conditions" + "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/overlay" ) @@ -146,8 +147,18 @@ func TestAMachineAwaitsAPushSinceItsOldestAssignmentSinceTheLastSend(t *testing. func TestALastSendIsReadFromTheSendAndTheReport(t *testing.T) { open := aMesh(t) ctx := t.Context() - register(t, open, aKeeper()) - pushedBy(t, open, "laptop") + // What a machine was sent is read from the commits its assignments name (issue 1109): a module is carried by a send only + // when it was built from a commit, and a hand send on a machine with nothing assigned leaves no key to mark unknown. + if err := open.inventory.RegisterModule(ctx, aKeeper(), inventory.Source{BuiltFrom: "c-keeper"}); err != nil { + t.Fatal(err) + } + record, err := open.inventory.NodeByName(ctx, "laptop") + if err != nil { + t.Fatal(err) + } + if err := open.inventory.RecordSent(ctx, record.ID, "carried", map[string]string{overlay.Name: "c-overlay"}); err != nil { + t.Fatal(err) + } if _, err := assign(ctx, open, "laptop", "keeper"); err != nil { t.Fatal(err) } diff --git a/cmd/mesh-controller/awaiting_push_test.go b/cmd/mesh-controller/awaiting_push_test.go index 0ec32245..58efd557 100644 --- a/cmd/mesh-controller/awaiting_push_test.go +++ b/cmd/mesh-controller/awaiting_push_test.go @@ -46,7 +46,21 @@ func pushedBy(t *testing.T, open *stores, node string) string { t.Fatal(err) } digest := digestOf(body) - if err := open.inventory.RecordSent(ctx, record.ID, digest, declared.Builds); err != nil { + // A push records the build of each module it carries (plan.go, Allocating); a module is carried by a send only when it was + // built from a commit, for what a machine was sent is read from the commits its assignments name (issue 1109). + builds := declared.Builds + if builds == nil { + current, err := open.inventory.CurrentBuilds(ctx) + if err != nil { + t.Fatal(err) + } + names := make([]string, 0, len(plan.Modules)) + for _, m := range plan.Modules { + names = append(names, m.Module) + } + builds = carriedBuilds(names, nil, current, nil) + } + if err := open.inventory.RecordSent(ctx, record.ID, digest, builds); err != nil { t.Fatal(err) } return digest diff --git a/cmd/mesh-controller/data_integration_test.go b/cmd/mesh-controller/data_integration_test.go index c1142df8..4d8a9853 100644 --- a/cmd/mesh-controller/data_integration_test.go +++ b/cmd/mesh-controller/data_integration_test.go @@ -56,13 +56,19 @@ func TestNatsUnassigningIrreplaceableDataRetiresItAndAnEmptyReplacementIsUrgent( if _, err := inv.SeedSeats(ctx, catalogue.DefaultSeats()); err != nil { t.Fatal(err) } - register(t, open, catalogue.Manifest{Module: "keeper", Version: "1", - Claims: []catalogue.Claim{{Name: catalogue.BackupSeat, Scope: catalogue.ScopeNode, Serves: []string{"backed-up", "now", "restore"}}}}) + // Built from a commit, so that a send carries them: a holder is asked only of what the machine was sent (issue 1109). + built := inventory.Source{BuiltFrom: "c1"} + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "keeper", Version: "1", + Claims: []catalogue.Claim{{Name: catalogue.BackupSeat, Scope: catalogue.ScopeNode, Serves: []string{"backed-up", "now", "restore"}}}}, built); err != nil { + t.Fatal(err) + } house, err := catalogue.ParseManifest([]byte(houseManifest)) if err != nil { t.Fatal(err) } - register(t, open, house) + if err := inv.RegisterModule(ctx, house, built); err != nil { + t.Fatal(err) + } if _, err := assign(ctx, open, "laptop", "house"); err == nil { t.Fatal("irreplaceable data was assigned to a machine with nothing to back it up") } diff --git a/cmd/mesh-controller/merge_gate.go b/cmd/mesh-controller/merge_gate.go index ec61888b..854f39ef 100644 --- a/cmd/mesh-controller/merge_gate.go +++ b/cmd/mesh-controller/merge_gate.go @@ -1120,6 +1120,9 @@ func throwawayStores(ctx context.Context, admin string) (*stores, func(), error) drop() return nil, nil, err } + // The chain lives on the bus, and the gate has none: this mesh is raised for a moment, with a chain of its own in memory + // that no other reads (issue 1121). + open.inventory.UseThrowawayChain() for _, m := range provided { if err := open.inventory.Provide(ctx, m); err != nil { open.Close() diff --git a/cmd/mesh-controller/merge_on_the_bus_test.go b/cmd/mesh-controller/merge_on_the_bus_test.go index df3825e2..3cfc5780 100644 --- a/cmd/mesh-controller/merge_on_the_bus_test.go +++ b/cmd/mesh-controller/merge_on_the_bus_test.go @@ -50,9 +50,17 @@ func theMergePathOn(t *testing.T, open *stores) func(link.SourceMoved) { t.Cleanup(func() { stop() <-done + lookingAtBatches.Wait() setWindowClock(nil) windowLook.Store(nil) }) + // The forge's merges are taken from a consumer made when the server starts serving: a merge published before it exists + // is not delivered to it. With the chain read from the bus the set-up above is quick enough to lose that race, so the + // publisher waits for the consumer. + becomes(t, 10*time.Second, "the merges' consumer being made", func() bool { + _, err := js.Context().ConsumerInfo("EVENTS", broker.MergesConsumerName) + return err == nil + }) return func(m link.SourceMoved) { t.Helper() body, err := json.Marshal(m) diff --git a/cmd/mesh-controller/merge_window_bus.go b/cmd/mesh-controller/merge_window_bus.go index 5b4cb0e3..bd721841 100644 --- a/cmd/mesh-controller/merge_window_bus.go +++ b/cmd/mesh-controller/merge_window_bus.go @@ -145,9 +145,15 @@ func (w *busWindow) Forget(ctx context.Context, batch string) { } } +// lookingAtBatches counts the looks under way, so a test that ends the window's clock can wait for the last of them before the +// next test replaces what they read. +var lookingAtBatches sync.WaitGroup + // lookAtBatches is the look an expired window asks for: the plans held, as a merge heard holds them, and the batches // whose window closed cut (cutBatchesHeld). Waiting for the plans rather than giving up: an expiry is heard once. func lookAtBatches(ctx context.Context, open *stores) { + lookingAtBatches.Add(1) + defer lookingAtBatches.Done() release, err := open.inventory.HoldPlans(ctx, true) if err != nil { fmt.Printf("batches: a window closed and the plans could not be held for the look: %v\n", err) diff --git a/cmd/mesh-controller/pending_integration_test.go b/cmd/mesh-controller/pending_integration_test.go index 5bbf1287..ba6a82fe 100644 --- a/cmd/mesh-controller/pending_integration_test.go +++ b/cmd/mesh-controller/pending_integration_test.go @@ -542,23 +542,26 @@ func TestAPendingAssignmentNeverWaitsSilently(t *testing.T) { t.Fatalf("a module registered meanwhile: %+v", p) } - // Ended rows are deleted after KeptFor (review point 7); open ones never, nor one whose condition is - // still open (second review, bug B). + // Ended waits are not deleted, for the chain removes no key (ADR 0408 section 4): after KeptFor they are no longer listed, open + // ones never are left out, nor one whose condition is still open (second review, bug B). if _, err := inv.RecordPending(ctx, inventory.PendingAssignment{Node: "anchor", Module: "lost", Build: "build-lost"}); err != nil { t.Fatal(err) } - said := settlePending(ctx, open, time.Now().Add(inventory.KeptFor+time.Hour)) - rows, err := inv.Pending(ctx, time.Unix(0, 0)) + later := time.Now().Add(inventory.KeptFor + time.Hour) + settlePending(ctx, open, later) + rows, err := inv.Pending(ctx, later.Add(-inventory.KeptFor)) if err != nil { t.Fatal(err) } for _, r := range rows { if !r.Open() && (r.Raised == nil || r.Cleared != nil) { - t.Fatalf("an ended pending assignment with no open condition outlived %s: %+v (%v)", inventory.KeptFor, r, said) + t.Fatalf("an ended pending assignment with no open condition is listed past %s: %+v", inventory.KeptFor, r) } } - if !strings.Contains(strings.Join(said, "\n"), "deleted 1 pending assignment(s)") { - t.Fatalf("the applied row was not deleted: %v", said) + for _, r := range rows { + if r.Module == "heard" { + t.Fatalf("an applied wait is listed past %s: %+v", inventory.KeptFor, r) + } } } diff --git a/cmd/mesh-controller/release.go b/cmd/mesh-controller/release.go index f7939794..bd7604c7 100644 --- a/cmd/mesh-controller/release.go +++ b/cmd/mesh-controller/release.go @@ -1589,7 +1589,9 @@ func grantsOnlyIn(ctx context.Context, open *stores, holder string, plan catalog var other []string listID := "" for _, m := range plan.Modules { - if _, was := sent[m.Module]; !was { + // A module no build is recorded for is carried with no commit and names none on its key (issue 1109), so it is "sent" + // only by the comparison of the summaries below; a module with a commit that was never sent is a new assignment. + if _, was := sent[m.Module]; !was && declared.Builds[m.Module] != "" { other = append(other, m.Module+" newly assigned") } if m.BusUsers != "" && m.ClaimsSeat(catalogue.BrokerSeat) { diff --git a/internal/inventory/chain.go b/internal/inventory/chain.go index a6e39ab3..1a0fbd7e 100644 --- a/internal/inventory/chain.go +++ b/internal/inventory/chain.go @@ -71,6 +71,14 @@ func (i *Inventory) UseChain(c *chain.Chain) { } } +// UseThrowawayChain gives the inventory a chain in memory that is its own, laid out already: for a mesh raised for a moment +// beside the real one, as the merge gate raises the facts snapshot's mesh in a throwaway store (issue 1121). Nothing of it +// reaches the bus. +func (i *Inventory) UseThrowawayChain() { + i.UseChain(chain.InMemory(nil)) + i.markLaidOut() +} + // attached is the chain this inventory has, nil when none. func (i *Inventory) attached() *chain.Chain { i.state.mu.Lock() diff --git a/internal/inventory/chain_pending_test.go b/internal/inventory/chain_pending_test.go index e9841507..3d48ba19 100644 --- a/internal/inventory/chain_pending_test.go +++ b/internal/inventory/chain_pending_test.go @@ -57,7 +57,8 @@ func TestAPendingAssignmentIsAKeyOfTheAssignmentsBucket(t *testing.T) { if known, _ := inv.PendingKnown(ctx, []string{p.ID, next.ID}); !known[p.ID] || !known[next.ID] { t.Fatalf("known %v: the key holds the newer one and carries the unanswered earlier one", known) } - if history, _ := inv.chain.Assignments.History(ctx, "anchor.sensors"); len(history) != 2 { + // The first wait, its settling and the new wait: three values, the earlier wait still in the history. + if history, _ := inv.chain.Assignments.History(ctx, "anchor.sensors"); len(history) != 3 { t.Fatalf("the earlier wait is in the key's history: %d values", len(history)) } }