diff --git a/cmd/mesh-controller/record_mirrors.go b/cmd/mesh-controller/record_mirrors.go index b1af389f..6bd907f7 100644 --- a/cmd/mesh-controller/record_mirrors.go +++ b/cmd/mesh-controller/record_mirrors.go @@ -44,7 +44,7 @@ func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats. return e } sources := map[string]func(context.Context) (map[string][]byte, error){ - recordmirror.Assignments.Name: inv.AssignmentsMirror, + recordmirror.Assignments.Name: inv.AssignmentsMirrorToWrite, recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror, recordmirror.Predefined.Name: inv.PredefinedMirror, } diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go index 67f9191b..958924e0 100644 --- a/internal/inventory/chain_assignments.go +++ b/internal/inventory/chain_assignments.go @@ -285,15 +285,13 @@ func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed i return written, removed, err } cur, has := have[key] - standing := has && cur.Value.Stands() - if standing && sameJSON(cur.Value.Provisioning, next.Provisioning) && cur.Value.RequiredBy == next.RequiredBy { + switch keepStep(has, cur.Value, next) { + case keepSkip: continue - } - switch { - case !has: + case keepCreate: next.Assigned = time.Time{} _, err = c.Assignments.Create(ctx, key, next) - case !standing: + case keepFresh: // A value that ended, pulled in again: a fresh one over it, the ended one staying in history. next.Assigned = time.Time{} _, err = c.Assignments.Update(ctx, key, next, cur.Revision) @@ -367,3 +365,32 @@ func (i *Inventory) KeptAssignments(ctx context.Context) (map[string]KeptValue, } return out, true, nil } + +// What a pass that keeps the plans does with one key. +const ( + keepSkip = "skip" + keepCreate = "create" + keepFresh = "fresh" + keepField = "field" +) + +// keepStep decides what the pass does with a key from what it wants there (read from the assignments) and what the bucket +// holds (read a moment later). **It never makes or revives an assignment**: that is Assign's, and an unassignment between the +// two reads leaves a key that ended while the pass still wants it as an assignment, which would come back as a value +// pulled in by another. Only a key with no time assigned, on both sides, is made or made again here. +func keepStep(has bool, cur, next chain.Assignment) string { + switch { + case !has && !next.Assigned.IsZero(): + return keepSkip // assigned since the bucket was read: its own write is not here yet + case !has: + return keepCreate + case !cur.Stands(): + if !cur.Assigned.IsZero() || !next.Assigned.IsZero() { + return keepSkip // an assignment that ended is not pulled in again by a plan + } + return keepFresh + case sameJSON(cur.Provisioning, next.Provisioning) && cur.RequiredBy == next.RequiredBy: + return keepSkip + } + return keepField +} diff --git a/internal/inventory/chain_assignments_key_test.go b/internal/inventory/chain_assignments_key_test.go index 47112e82..62e87cc7 100644 --- a/internal/inventory/chain_assignments_key_test.go +++ b/internal/inventory/chain_assignments_key_test.go @@ -1,8 +1,13 @@ package inventory import ( + "context" "encoding/json" + "errors" "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" ) // **An assignment's key is `.`, as the bucket has always had it** (ADR 0408 §3): its readers parse nothing @@ -35,3 +40,42 @@ func TestAnAbsentPlanAndANullOneAreTheSame(t *testing.T) { } } } + +// **A pass that keeps the plans never brings back an assignment that ended** (issue 1093): it reads what it wants and what +// the bucket holds in two reads, and an unassignment between them leaves a key that ended while the pass still wants it +// assigned. +func TestAPassThatKeepsThePlansNeverRevivesAnEndedAssignment(t *testing.T) { + at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + plan := json.RawMessage(`{"files":[]}`) + assigned := chain.Assignment{Node: "a", Module: "m", Assigned: at, Provisioning: plan} + ended := chain.Assignment{Node: "a", Module: "m", Assigned: at, Ended: at.Add(time.Minute)} + pulledIn := chain.Assignment{Node: "a", Module: "p", RequiredBy: "m", Provisioning: plan} + pulledInEnded := chain.Assignment{Node: "a", Module: "p", Ended: at} + for name, tc := range map[string]struct { + has bool + cur, next chain.Assignment + want string + }{ + "the interleaving: wanted assigned, ended since": {true, ended, assigned, keepSkip}, + "wanted assigned, not in the bucket yet": {false, chain.Assignment{}, assigned, keepSkip}, + "a pulled-in module, not in the bucket": {false, chain.Assignment{}, pulledIn, keepCreate}, + "a pulled-in module that ended, pulled again": {true, pulledInEnded, pulledIn, keepFresh}, + "an assignment ended, a plan for it": {true, ended, chain.Assignment{Node: "a", Module: "m", RequiredBy: "x", Provisioning: plan}, keepSkip}, + "the plan is the same": {true, assigned, assigned, keepSkip}, + "the plan changed": {true, chain.Assignment{Node: "a", Module: "m", Assigned: at}, assigned, keepField}, + } { + if got := keepStep(tc.has, tc.cur, tc.next); got != tc.want { + t.Errorf("%s: %s, want %s", name, got, tc.want) + } + } +} + +// **The store's mirror of the assignments is refused to its writer once the assignments are moved** (ADR 0408 §4, issue +// 1093): asked at the moment of the write, so a tick that decided before the switch lays nothing after it. +func TestTheStoresMirrorOfTheAssignmentsIsRefusedToItsWriterOnceTheyAreMoved(t *testing.T) { + inv := &Inventory{moved: &movedKinds{kind: map[string]bool{KindAssignments: true}}} + inv.UseChain(chain.InMemory(nil)) + if _, err := inv.AssignmentsMirrorToWrite(context.Background()); !errors.Is(err, ErrAssignmentsMoved) { + t.Fatalf("the writer was given the store's rows: %v", err) + } +} diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go index a9afde08..d62363c2 100644 --- a/internal/inventory/chain_move.go +++ b/internal/inventory/chain_move.go @@ -1,6 +1,7 @@ package inventory import ( + "bytes" "context" "encoding/json" "errors" @@ -59,6 +60,10 @@ type mover struct { count func(ctx context.Context, tx pgx.Tx) (int, error) // lock keeps the tables from taking a row while the switch is recorded. lock string + // settle runs under that lock, with the rows read again: it makes the bucket hold what the rows say now. A row the old + // controller changed after the first pass is in the bucket before the kind is recorded as moved, for before the switch + // the table is the only truth. + settle func(ctx context.Context, c *chain.Chain, rows []moveRow) error } type moveRow struct { @@ -122,6 +127,9 @@ func (i *Inventory) movers() []mover { return n, err }, lock: `lock table build, build_request in share mode`, + settle: func(ctx context.Context, c *chain.Chain, rows []moveRow) error { + return settleBucket(ctx, c.Builds, rows) + }, }, { kind: KindRollouts, rows: func(ctx context.Context) ([]moveRow, int, error) { @@ -160,6 +168,9 @@ func (i *Inventory) movers() []mover { return n, err }, lock: `lock table release_plan, batched_merge in share mode`, + settle: func(ctx context.Context, c *chain.Chain, rows []moveRow) error { + return settleBucket(ctx, chain.RolloutsOf[Plan](c), rows) + }, }, { kind: KindAssignments, rows: func(ctx context.Context) ([]moveRow, int, error) { @@ -177,41 +188,38 @@ func (i *Inventory) movers() []mover { } return out, len(out), nil }, - create: func(ctx context.Context, c *chain.Chain, row moveRow) error { - want := row.value.(chain.Assignment) - // The mirror of the store kept a value under this key until now, with the provisioner's plan beside it: the - // plan stays, the record's links are laid in. A value that names its build already is a later word, and stays. - for try := 0; try < casTries; try++ { - cur, found, err := c.Assignments.Get(ctx, row.key) - if err != nil { - return err - } - if !found { - if _, err := c.Assignments.Create(ctx, row.key, want); !errors.Is(err, chain.ErrExists) { - return err - } - continue - } - if cur.Value.Build != "" || cur.Value.Commit != "" || !cur.Value.Since.IsZero() || (want.Build == "" && want.Commit == "") { - return chain.ErrExists - } - merged := want - merged.RequiredBy, merged.Provisioning = cur.Value.RequiredBy, cur.Value.Provisioning - if _, err := c.Assignments.Update(ctx, row.key, merged, cur.Revision); errors.Is(err, chain.ErrMoved) { - continue - } else { - return err - } - } - return fmt.Errorf("the assignment %s kept moving under the write", row.key) - }, - keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Assignments.Keys(ctx) }, + create: createAssignment, + keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Assignments.Keys(ctx) }, count: func(ctx context.Context, tx pgx.Tx) (int, error) { var n int err := tx.QueryRow(ctx, `select count(*) from assignment`).Scan(&n) return n, err }, lock: `lock table assignment in share mode`, + settle: func(ctx context.Context, c *chain.Chain, rows []moveRow) error { + have := make(map[string]bool, len(rows)) + for _, r := range rows { + have[r.key] = true + if err := settleAssignment(ctx, c, r); err != nil { + return err + } + } + // An assignment the store no longer holds ended since the first pass: the bucket says so. + all, err := c.Assignments.All(ctx) + if err != nil { + return err + } + for key, k := range all { + if have[key] || k.Value.Assigned.IsZero() || !k.Value.Stands() { + continue + } + k.Value.Ended = time.Now().UTC() + if _, err := c.Assignments.Update(ctx, key, k.Value, k.Revision); err != nil && !errors.Is(err, chain.ErrMoved) { + return err + } + } + return nil + }, }} } @@ -444,6 +452,17 @@ func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool if n != r.TableRows { return false, nil } + // The rows read again under the lock: nothing writes them now, so what the bucket holds is made what they say. + rows, tableRows, err := m.rows(ctx) + if err != nil { + return false, err + } + if tableRows != r.TableRows { + return false, nil + } + if err := m.settle(ctx, i.attached(), rows); err != nil { + return false, fmt.Errorf("the %s changed since they were first written, and the bucket cannot be brought to them: %w", m.kind, err) + } counts, _ := json.Marshal(map[string]int{m.kind: r.Rows}) largest, _ := json.Marshal(map[string]int{m.kind: r.Largest}) if _, err := tx.Exec(ctx, @@ -527,3 +546,106 @@ func (i *Inventory) allRequestRows(ctx context.Context) ([]BuildRequest, error) } return out, rows.Err() } + +// createAssignment writes an assignment's row under its key; chain.ErrExists when the key holds a value that stays. +func createAssignment(ctx context.Context, c *chain.Chain, row moveRow) error { + want := row.value.(chain.Assignment) + // The mirror of the store kept a value under this key until now, with the provisioner's plan beside it: the + // plan stays, the record's links are laid in. A value that names its build already is a later word, and stays. + for try := 0; try < casTries; try++ { + cur, found, err := c.Assignments.Get(ctx, row.key) + if err != nil { + return err + } + if !found { + if _, err := c.Assignments.Create(ctx, row.key, want); !errors.Is(err, chain.ErrExists) { + return err + } + continue + } + if cur.Value.Build != "" || cur.Value.Commit != "" || !cur.Value.Since.IsZero() || (want.Build == "" && want.Commit == "") { + return chain.ErrExists + } + merged := want + merged.RequiredBy, merged.Provisioning = cur.Value.RequiredBy, cur.Value.Provisioning + if _, err := c.Assignments.Update(ctx, row.key, merged, cur.Revision); errors.Is(err, chain.ErrMoved) { + continue + } else { + return err + } + } + return fmt.Errorf("the assignment %s kept moving under the write", row.key) +} + +// settleBucket makes a bucket of whole values hold each row's value, by compare-and-set: a key that is absent is made, one +// whose bytes differ from the row's is written over, and one that is the same is left. +func settleBucket[T any](ctx context.Context, b chain.Bucket[T], rows []moveRow) error { + for _, r := range rows { + want, err := json.Marshal(r.value) + if err != nil { + return err + } + settled := false + for try := 0; try < casTries && !settled; try++ { + e, found, err := b.GetRaw(ctx, r.key) + if err != nil { + return err + } + switch { + case !found: + _, err = b.CreateRaw(ctx, r.key, want) + if errors.Is(err, chain.ErrExists) { + continue + } + case bytes.Equal(e.Value, want): + settled = true + continue + default: + _, err = b.UpdateRaw(ctx, r.key, want, e.Revision) + if errors.Is(err, chain.ErrMoved) { + continue + } + } + if err != nil { + return err + } + settled = true + } + if !settled { + return fmt.Errorf("%s kept moving under the write", r.key) + } + } + return nil +} + +// settleAssignment makes the value under an assignment's key what its row says, keeping the plan the mirror put beside it: +// before the switch the store is the only truth of the links, so a commit it changed since the first pass is written over. +func settleAssignment(ctx context.Context, c *chain.Chain, row moveRow) error { + want := row.value.(chain.Assignment) + for try := 0; try < casTries; try++ { + cur, found, err := c.Assignments.Get(ctx, row.key) + if err != nil { + return err + } + if !found { + if _, err := c.Assignments.Create(ctx, row.key, want); errors.Is(err, chain.ErrExists) { + continue + } else { + return err + } + } + merged := want + merged.RequiredBy, merged.Provisioning = cur.Value.RequiredBy, cur.Value.Provisioning + a, _ := json.Marshal(merged) + b, _ := json.Marshal(cur.Value) + if bytes.Equal(a, b) { + return nil + } + if _, err := c.Assignments.Update(ctx, row.key, merged, cur.Revision); errors.Is(err, chain.ErrMoved) { + continue + } else { + return err + } + } + return fmt.Errorf("the assignment %s kept moving under the write", row.key) +} diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go index 56ddf33a..2648eff8 100644 --- a/internal/inventory/chain_move_test.go +++ b/internal/inventory/chain_move_test.go @@ -9,6 +9,7 @@ import ( "reflect" "slices" "strings" + "sync" "testing" "time" @@ -34,6 +35,31 @@ func aBuildRow(t *testing.T, inv *Inventory, id, module, commit, failed string) id, module, commit, failed) } +// reportOf is the report of one kind among those a move made, which are one for each kind in the order they are moved. +func reportOf(t *testing.T, reports []MoveReport, kind string) MoveReport { + t.Helper() + for _, r := range reports { + if r.Kind == kind { + return r + } + } + t.Fatalf("no report of %s among %+v", kind, reports) + return MoveReport{} +} + +// allAlready says every kind was moved by an earlier start: the second run writes nothing, and reports one for each kind. +func allAlready(t *testing.T, reports []MoveReport) { + t.Helper() + if len(reports) != 3 { + t.Fatalf("the move reported %d kinds, want 3: %+v", len(reports), reports) + } + for _, r := range reports { + if !r.Already || r.Written != 0 { + t.Fatalf("the second move of %s %+v", r.Kind, r) + } + } +} + func rowsInTheTable(t *testing.T, inv *Inventory, table string) string { t.Helper() rows := inv.QueryForTest(t, `select count(*) from `+table) @@ -63,8 +89,8 @@ func TestTheMoveCopiesTheBuildRowsAndAgainWritesNothing(t *testing.T) { if err != nil { t.Fatal(err) } - if len(reports) != 1 || reports[0].Kind != KindBuilds || reports[0].Rows != 3 || reports[0].Written != 3 || reports[0].Largest == 0 { - t.Fatalf("the move reported %+v", reports) + if b := reportOf(t, reports, KindBuilds); b.Rows != 3 || b.Written != 3 || b.Largest == 0 { + t.Fatalf("the move of the builds reported %+v", reports) } moved, err := inv.ChainMoved(ctx) if err != nil || moved[KindBuilds] != 3 { @@ -94,9 +120,10 @@ func TestTheMoveCopiesTheBuildRowsAndAgainWritesNothing(t *testing.T) { // A second run finds the kind moved and writes nothing: every key still has the one value it was made with. again, err := inv.MoveChain(ctx, nil) - if err != nil || len(again) != 1 || !again[0].Already || again[0].Written != 0 { - t.Fatalf("the second move %+v: %v", again, err) + if err != nil { + t.Fatal(err) } + allAlready(t, again) for _, id := range []string{"build-1", "build-2", "build-3", "build-4"} { if history, err := inv.chain.Builds.History(ctx, chainBuildKey(id)); err != nil || len(history) != 1 { t.Fatalf("%s has %d values: %v", id, len(history), err) @@ -122,8 +149,8 @@ func TestAControllerRestartedMidMoveFinishesItAndOverwritesNothing(t *testing.T) if err != nil { t.Fatal(err) } - if reports[0].Written != 2 || reports[0].Present != 1 || reports[0].Rows != 3 { - t.Fatalf("the move reported %+v", reports[0]) + if b := reportOf(t, reports, KindBuilds); b.Written != 2 || b.Present != 1 || b.Rows != 3 { + t.Fatalf("the move reported %+v", b) } got, _, _ := inv.chain.Builds.Get(ctx, "build-1") if !reflect.DeepEqual(got.Value, newer) { @@ -387,8 +414,8 @@ func TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain(t *testin if err != nil { t.Fatal(err) } - if reports[0].Rows != 3 || reports[0].TableRows != 4 { - t.Fatalf("the move reported %+v", reports[0]) + if b := reportOf(t, reports, KindBuilds); b.Rows != 3 || b.TableRows != 4 { + t.Fatalf("the move reported %+v", b) } one, _, _ := inv.chain.Builds.Get(ctx, "build-1") if one.Value.State != "" || one.Value.Request == nil || one.Value.Request.For != "merge" || one.Value.Module != "notes" { @@ -522,9 +549,11 @@ func TestTheMoveCarriesThePlansAndTheirMergesAndTheyReadTheSame(t *testing.T) { if n := rowsInTheTable(t, inv, "release_plan"); n != "2" { t.Fatalf("the table took a plan after the move: %s", n) } - if again, err := inv.MoveChain(ctx, nil); err != nil || len(again) != 2 || !again[0].Already || !again[1].Already { - t.Fatalf("the second move %+v: %v", again, err) + again, err := inv.MoveChain(ctx, nil) + if err != nil { + t.Fatal(err) } + allAlready(t, again) } // **An assignment is a key of the assignments bucket; an unassignment writes `ended` and keeps the key and its values; a @@ -779,3 +808,79 @@ func TestTheMoveCarriesTheAssignmentsAndWhatEachMachineWasSent(t *testing.T) { t.Fatalf("assigned on laptop: %v", on) } } + +// hooked runs a function the first time a key is made in a bucket: the old controller writing the store between the move's +// first pass and its switch. +type hooked struct { + chain.Backend + once *sync.Once + then func() +} + +func (h hooked) Create(ctx context.Context, key string, value []byte) (uint64, error) { + rev, err := h.Backend.Create(ctx, key, value) + h.once.Do(h.then) + return rev, err +} + +// **A row the old controller changes after the move's first pass is in the bucket when the kind is recorded as moved** +// (ADR 0408 §8, issue 981 note 7 Q2): before the switch the store is the only truth, so under the lock the move reads the +// rows again and writes over every key that differs — a plan's state and a build's failure — and ends an assignment the +// store dropped. The count alone would have kept the older value and refused the next save of the plan. +func TestARowChangedAfterTheFirstPassIsInTheBucketWhenTheKindIsRecordedMoved(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + for _, n := range []string{"anchor"} { + if _, err := inv.AddNode(ctx, n); err != nil { + t.Fatal(err) + } + } + for _, m := range []string{"notes", "wiki"} { + if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: "1"}, Source{}); err != nil { + t.Fatal(err) + } + } + for _, m := range []string{"notes", "wiki"} { + if _, err := inv.Assign(ctx, "anchor", m); err != nil { + t.Fatal(err) + } + } + p := aBatch("plan-1", time.Now().Add(-time.Hour).UTC()) + p.State, p.Tiers = PlanBuilding, [][]string{{"notes"}} + if err := inv.SavePlan(ctx, p); err != nil { + t.Fatal(err) + } + // Between a kind's first pass and its switch the old controller saves the plan again, fails the build and unassigns wiki: + // each is done when the first key of its kind is made, so the first pass has read the rows as they were. + after := func(sql string) hooked { + return hooked{chain.NewMemory(), &sync.Once{}, func() { inv.ExecForTest(t, sql) }} + } + inv.UseChain(chain.New( + after(`update build set failed = 'it broke later' where id = 'build-1'`), + after(`update release_plan set state = 'rolling', revision = revision + 1, note = 'the old controller was here' where id = 'plan-1'`), + after(`delete from assignment where module = 'wiki'`), + func(context.Context) error { return nil })) + if _, err := inv.MoveChain(ctx, nil); err != nil { + t.Fatal(err) + } + now, err := inv.PlanByID(ctx, "plan-1") + if err != nil { + t.Fatal(err) + } + if now.State != PlanRolling || now.Note != "the old controller was here" || now.Revision != 2 { + t.Fatalf("the bucket holds a plan from before the change: %s %q revision %d", now.State, now.Note, now.Revision) + } + if b, found, _ := inv.BuildByID(ctx, "build-1"); !found || b.Failed != "it broke later" { + t.Fatalf("the bucket holds the build as it was: %+v", b) + } + if on, _ := inv.Assigned(ctx, "anchor"); !slices.Equal(on, []string{"notes"}) { + t.Fatalf("anchor is assigned %v, though the store dropped wiki", on) + } + // The plan can be saved at the revision the store had. + now.Note = "saved after the move" + if err := inv.SavePlan(ctx, &now); err != nil { + t.Fatalf("the next save of the plan is refused: %v", err) + } +} diff --git a/internal/inventory/mirrors.go b/internal/inventory/mirrors.go index 3f35f767..fadab025 100644 --- a/internal/inventory/mirrors.go +++ b/internal/inventory/mirrors.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "time" ) @@ -69,6 +70,21 @@ type DeclaredMirrored struct { var DeclaredMirroredFields = []string{"kind", "default", "why", "type", "description", "enum", "minimum", "maximum", "pattern", "trusted", "predefined", "required", "runtime"} +// ErrAssignmentsMoved is a write of the store's mirror of the assignments, refused because the assignments are keys of the bus. +var ErrAssignmentsMoved = errors.New("the assignments are keys of the bus, and the store's rows are no longer what they say") + +// AssignmentsMirrorToWrite is AssignmentsMirror for the mirror's writer, which lays what it returns over the bucket and +// ends the keys it does not return: **refused once the assignments are moved** (ADR 0408 §4), asked at the moment of the +// write, so a tick that decided before the switch cannot lay mirror-shaped values over the links after it. +func (i *Inventory) AssignmentsMirrorToWrite(ctx context.Context) (map[string][]byte, error) { + if _, on, err := i.ChainReads(ctx, KindAssignments); err != nil { + return nil, err + } else if on { + return nil, ErrAssignmentsMoved + } + return i.AssignmentsMirror(ctx) +} + // AssignmentsMirror is every assignment's mirrored value, keyed `.`, with the provisioner's plan of // each assignment of a module handed to it, and a value of its own for each module handed to it on a node that // nothing assigned there, only pulled in by another (issue 765). diff --git a/modules/route-proxy/go.mod b/modules/route-proxy/go.mod index 54bca71c..a9826b99 100644 --- a/modules/route-proxy/go.mod +++ b/modules/route-proxy/go.mod @@ -2,4 +2,4 @@ module route-proxy go 1.25.0 -require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018003858-f185aedbf5d6 diff --git a/modules/route-proxy/go.sum b/modules/route-proxy/go.sum index 63b4da41..1a769534 100644 --- a/modules/route-proxy/go.sum +++ b/modules/route-proxy/go.sum @@ -1,2 +1,4 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 h1:EK84so3Y8JdFT3BHrzj9cPQQrDUPa/d+IdOGnerlmEI= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018003858-f185aedbf5d6 h1:Sv1Qzix+U1RpaUkw5IbB0IJEyZ+FqgzHmbZrCZkcWf0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018003858-f185aedbf5d6/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= diff --git a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go index 33bf61f7..4adee9e9 100644 --- a/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go +++ b/modules/route-proxy/vendor/git.novox.be/novox/mesh-sdk/go/provider/provider.go @@ -335,6 +335,9 @@ func (p *Provider) pass(ctx context.Context, assignments, entries map[string][]b var out []Outcome unknown := map[string]bool{} // consumers whose assignment was skipped: their entries are not judged for _, key := range sortedKeys(assignments) { + if !Standing(assignments[key]) { + continue // ended or pending: read as a removed key (issue 1098) + } var a planned if err := json.Unmarshal(assignments[key], &a); err != nil || a.Node+"."+a.Module != key { o := Outcome{Key: key, Status: Failed, Why: fmt.Sprintf("the assignment %s is not the controller's, so "+ @@ -744,3 +747,23 @@ func (b stateBus) Delete(_ context.Context, bucket, key string) error { } return s.Delete(key) } + +// Standing is whether an assignment's value is a standing assignment: false when `ended` is set (the module was +// taken off its node, novox/hq ADR 0408 clause 4) or `state` is not empty (a pending one). Such a value is read +// exactly as a removed key. Every other field, known or not, changes nothing (issue 1098). +func Standing(value []byte) bool { + var v struct { + Ended json.RawMessage `json:"ended"` + State json.RawMessage `json:"state"` + } + if json.Unmarshal(value, &v) != nil { + return true // not decided here: the parse that follows refuses it + } + return !fieldSet(v.Ended) && !fieldSet(v.State) +} + +// fieldSet is whether a field carries something: not absent, null or empty. +func fieldSet(f json.RawMessage) bool { + s := strings.TrimSpace(string(f)) + return s != "" && s != "null" && s != `""` +} diff --git a/modules/route-proxy/vendor/modules.txt b/modules/route-proxy/vendor/modules.txt index abae1a04..a4b722ac 100644 --- a/modules/route-proxy/vendor/modules.txt +++ b/modules/route-proxy/vendor/modules.txt @@ -1,4 +1,4 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261016174426-a4e7f480b3b0 +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018003858-f185aedbf5d6 ## explicit; go 1.22 git.novox.be/novox/mesh-sdk/go git.novox.be/novox/mesh-sdk/go/provider diff --git a/testdata/store-tests b/testdata/store-tests index 0717a993..91c80206 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -154,3 +154,4 @@ ./internal/inventory TestTheMoveCarriesTheAssignmentsAndWhatEachMachineWasSent ./cmd/mesh-controller TestOneReadOfAnAssignmentNamesItsCommitBuildAndRolloutAndOneReadOfTheRolloutListsItsAssignments ./cmd/mesh-controller TestASendOfWhatAMachineRunsAlreadyChangesNoLinkOfItsAssignment +./internal/inventory TestARowChangedAfterTheFirstPassIsInTheBucketWhenTheKindIsRecordedMoved