diff --git a/cmd/mesh-controller/chain.go b/cmd/mesh-controller/chain.go index f54e6258..83239d20 100644 --- a/cmd/mesh-controller/chain.go +++ b/cmd/mesh-controller/chain.go @@ -7,6 +7,7 @@ import ( "time" "github.com/novox/mesh-controller/internal/chain" + "github.com/novox/mesh-controller/internal/conditions" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) @@ -43,6 +44,8 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link. inv.UseChain(c) // And the requests the move of delivery 1 found never handed over, asked again once (askAgain), once it is serving. go askAgainSoon(ctx, inv) + // And what delivery 1 recorded of what each machine was sent, copied onto the keys of the assignments (CopySentBuilds). + go copySentBuildsSoon(ctx, inv) return nil } @@ -87,3 +90,43 @@ func askAgainSoon(ctx context.Context, inv *inventory.Inventory) { askAgain(ctx, inv) } } + +// kindSentBuildsNotCopied is the condition of the one-time copy of node.sent_builds onto the assignments' keys that has not +// succeeded: until it has, a module an assignment pulled in may be missing from what a machine is read to run. +const kindSentBuildsNotCopied = "sent-builds-not-copied" + +// copySentBuildsSoon copies what the store recorded of what each machine was sent onto the keys of the assignments, as soon +// as this controller holds the lease (the chain refuses a write before), and tries again every half minute while it fails, +// each time it fails saying so as an urgent condition until it has worked. It is idempotent: CopySentBuilds overwrites no key +// that names a commit and empties each column it has copied. +func copySentBuildsSoon(ctx context.Context, inv *inventory.Inventory) { + failed := false + for { + _, err := inv.CopySentBuilds(ctx) + o := conditions.Observation{Scope: conditions.ScopeBuild, ID: kindSentBuildsNotCopied, Token: kindSentBuildsNotCopied, + Kind: kindSentBuildsNotCopied, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, + Source: "chain", Summary: "what each machine was sent is not copied onto its assignments: "} + if err != nil { + fmt.Printf("chain: what the machines were sent is not copied onto the assignments: %v\n", err) + if keeper := conditionsFrom; keeper != nil && !failed { + o.Summary += err.Error() + if _, oerr := keeper.Observe(ctx, o); oerr == nil { + failed = true + } + } + } else { + if keeper := conditionsFrom; keeper != nil && failed { + _, _ = keeper.Clear(ctx, o.Key(), "copied") + } + return + } + select { + case <-ctx.Done(): + return + case <-time.After(chainMoveRetry): + } + } +} + +// chainMoveRetry is how long the copy waits before it tries again. +const chainMoveRetry = 30 * time.Second diff --git a/cmd/mesh-controller/chain_no_table_integration_test.go b/cmd/mesh-controller/chain_no_table_integration_test.go index 0bb3a10a..20181348 100644 --- a/cmd/mesh-controller/chain_no_table_integration_test.go +++ b/cmd/mesh-controller/chain_no_table_integration_test.go @@ -13,8 +13,7 @@ import ( // **Acceptance 12 of design 47: no table in the controller's store holds builds, rollouts or assignments** (novox/hq ADR // 0408, issue 1111). A delivery is driven through the live trigger — a merge heard, its batch cut into a plan, the plan's // build asked and answered, the assignment made — until a build, a rollout and an assignment exist; they are read from the -// chain, and then the store's own catalog is read: none of the tables of the chain is in it, nor the column that held what -// a machine was sent, nor the function that raised the generation from the assignment table; the table that stays, stays. +// chain, and then the store's own catalog is read: none of the tables of the chain is in it, nor the function that raised the generation from the assignment table; the table that stays, stays. func TestNoTableOfTheControllersStoreHoldsBuildsRolloutsOrAssignments(t *testing.T) { open := windowed(t) everyHeadWaits(t) @@ -80,9 +79,6 @@ func TestNoTableOfTheControllersStoreHoldsBuildsRolloutsOrAssignments(t *testing t.Errorf("the table %s, which stays, is gone", table) } } - if rows := inv.QueryForTest(t, `select count(*) from information_schema.columns where table_schema = 'public' and column_name = 'sent_builds'`); rows[0][0] != "0" { - t.Errorf("a column sent_builds is in the store: %v", rows) - } if rows := inv.QueryForTest(t, `select count(*) from pg_proc where proname = 'raise_assignment_generation'`); rows[0][0] != "0" { t.Errorf("the function raise_assignment_generation is in the store: %v", rows) } diff --git a/internal/inventory/chain_drop_test.go b/internal/inventory/chain_drop_test.go index 80c20517..12a9b3ea 100644 --- a/internal/inventory/chain_drop_test.go +++ b/internal/inventory/chain_drop_test.go @@ -80,8 +80,8 @@ func TestMigration0105RefusesAStoreWhoseRowsWereNotMovedAndDropsTheTablesOfAMove t.Errorf("the table %s is gone", stays) } } - if rows := inv.QueryForTest(t, `select count(*) from information_schema.columns where table_name = 'node' and column_name = 'sent_builds'`); rows[0][0] != "0" { - t.Errorf("node.sent_builds is still a column") + if rows := inv.QueryForTest(t, `select count(*) from information_schema.columns where table_name = 'node' and column_name = 'sent_builds'`); rows[0][0] != "1" { + t.Errorf("node.sent_builds, which stays as the source of the copy, is gone") } if rows := inv.QueryForTest(t, `select count(*) from pg_proc where proname = 'raise_assignment_generation'`); rows[0][0] != "0" { t.Errorf("the function raise_assignment_generation is still in the store") diff --git a/internal/inventory/chain_links.go b/internal/inventory/chain_links.go index 69458990..56ec03b4 100644 --- a/internal/inventory/chain_links.go +++ b/internal/inventory/chain_links.go @@ -2,8 +2,11 @@ package inventory import ( "context" + "encoding/json" "errors" "fmt" + "maps" + "slices" "sort" "time" @@ -254,3 +257,106 @@ func (i *Inventory) recordRun(ctx context.Context, c *chain.Chain, node, module, } return fmt.Errorf("the assignment kept moving under the write") } + +// CopySentBuilds copies what delivery 1 recorded in node.sent_builds onto the keys of the assignments, once, at the start of +// the controller that holds the lease and has attached the chain (issue 1109, review round 1): delivery 1 wrote no commit onto +// the key of a module an assignment pulled in, so without it those modules would be missing from SentBuilds although the +// machine is known to have been sent. A key that names a commit is never overwritten, so it is idempotent; once a node's keys +// are written its column is emptied, so a later restart cannot lay a stale commit over a key a later send cleared or ended. +// It answers how many keys it wrote; an error is raised by the caller as a condition. +func (i *Inventory) CopySentBuilds(ctx context.Context) (int, error) { + c, err := i.theChain(ctx) + if err != nil { + return 0, err + } + rows, err := i.store.Pool().Query(ctx, `select name, sent_builds from node where sent_builds is not null`) + if err != nil { + return 0, err + } + type kept struct { + name string + sent map[string]string + } + var all []kept + for rows.Next() { + var name string + var raw []byte + if err := rows.Scan(&name, &raw); err != nil { + rows.Close() + return 0, err + } + sent := map[string]string{} + if err := json.Unmarshal(raw, &sent); err != nil { + rows.Close() + return 0, fmt.Errorf("what %s was sent, in the store, cannot be read: %w", name, err) + } + all = append(all, kept{name, sent}) + } + rows.Close() + if err := rows.Err(); err != nil { + return 0, err + } + now := time.Now().UTC() + for _, k := range all { + if err := i.copySentBuildsOf(ctx, c, k.name, k.sent, now); err != nil { + return 0, fmt.Errorf("what %s was sent cannot be copied onto its assignments: %w", k.name, err) + } + if _, err := i.store.Pool().Exec(ctx, `update node set sent_builds = null where name = $1`, k.name); err != nil { + return 0, err + } + } + return len(all), nil +} + +// copySentBuildsOf is CopySentBuilds for one node, over a chain: for each module the node was sent, a standing key that +// names no commit gets it (and the build, when one is known; no rollout), and where no key stands a standing pulled-in key is +// made. A key with a commit, a hand send's unknown and a wait for a build are left as they are. +func (i *Inventory) copySentBuildsOf(ctx context.Context, c *chain.Chain, node string, sent map[string]string, at time.Time) error { + for _, module := range slices.Sorted(maps.Keys(sent)) { + commit := sent[module] + if commit == "" { + continue + } + key := assignmentKey(node, module) + build, err := i.chainBuildOf(ctx, c, module, commit) + if err != nil { + return err + } + for try := 0; ; try++ { + if try == casTries { + return fmt.Errorf("the assignment %s kept moving under the write", key) + } + cur, found, err := c.Assignments.Get(ctx, key) + if err != nil { + return err + } + if found && (cur.Value.State != "" || cur.Value.Commit != "" || !cur.Value.SentUnknown.IsZero()) { + break + } + next := cur.Value + if !found || !cur.Value.Stands() { + next = chain.Assignment{Node: node, Module: module, PulledIn: true, Since: at} + } + next.Commit, next.Build = commit, build + if next.Since.IsZero() { + next.Since = at + } + if !found { + _, err = c.Assignments.Create(ctx, key, next) + if errors.Is(err, chain.ErrExists) { + continue + } + } else { + _, err = c.Assignments.Update(ctx, key, next, cur.Revision) + if errors.Is(err, chain.ErrMoved) { + continue + } + } + if err != nil { + return err + } + break + } + } + return nil +} diff --git a/internal/inventory/chain_sent_copy_test.go b/internal/inventory/chain_sent_copy_test.go new file mode 100644 index 00000000..49bf9474 --- /dev/null +++ b/internal/inventory/chain_sent_copy_test.go @@ -0,0 +1,90 @@ +package inventory + +import ( + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// **What delivery 1 recorded in node.sent_builds is copied onto the keys of the assignments, once** (issue 1109, review +// round 1): a module pulled in by an assignment had no commit on its key, so it would fall out of SentBuilds. The copy +// creates a standing pulled-in key where none stands, fills a standing key that names no commit (commit, and the build +// when one is known, no rollout), and overwrites nothing: a key with a commit, a hand send's unknown, a wait for a build. +func TestWhatTheColumnRecordedIsCopiedOntoTheKeysAndOverwritesNothing(t *testing.T) { + inv, c := anInventoryOnTheChain(t) + ctx := t.Context() + at := time.Date(2026, 10, 18, 4, 0, 0, 0, time.UTC) + if _, err := c.Builds.Create(ctx, "b-wg", chain.Build{ID: "b-wg", Module: "mesh-wireguard", Commit: "c-wg"}); err != nil { + t.Fatal(err) + } + put := func(module string, a chain.Assignment) { + a.Node, a.Module = "laptop", module + if _, err := c.Assignments.Create(ctx, assignmentKey("laptop", module), a); err != nil { + t.Fatal(err) + } + } + put("notes", chain.Assignment{Assigned: at}) // assigned, no commit yet + put("web", chain.Assignment{Assigned: at, Commit: "c-web-new", Build: "b1"}) // already has one + put("hand", chain.Assignment{Assigned: at, SentUnknown: at}) // a hand send since + put("waits", chain.Assignment{Assigned: at, State: chain.StatePending}) // a wait for a build + put("ended", chain.Assignment{Assigned: at, Ended: at}) // unassigned + sent := map[string]string{"notes": "c-notes", "web": "c-web-old", "mesh-wireguard": "c-wg", "hand": "c-hand", + "waits": "c-waits", "ended": "c-ended"} + + for pass := 0; pass < 2; pass++ { // and the second pass changes nothing + if err := inv.copySentBuildsOf(ctx, c, "laptop", sent, at); err != nil { + t.Fatal(err) + } + get := func(module string) chain.Assignment { + t.Helper() + k, found, err := c.Assignments.Get(ctx, assignmentKey("laptop", module)) + if err != nil || !found { + t.Fatalf("%s: %v %v", module, found, err) + } + return k.Value + } + if a := get("notes"); a.Commit != "c-notes" || a.Rollout != "" || !a.Stands() || a.PulledIn { + t.Errorf("notes: %+v", a) + } + if a := get("web"); a.Commit != "c-web-new" || a.Build != "b1" { + t.Errorf("a key with a commit was overwritten: %+v", a) + } + if a := get("hand"); a.Commit != "" || a.SentUnknown.IsZero() { + t.Errorf("a hand send's unknown was overwritten: %+v", a) + } + if a := get("waits"); a.Commit != "" || a.State != chain.StatePending { + t.Errorf("a wait was overwritten: %+v", a) + } + if a := get("mesh-wireguard"); a.Commit != "c-wg" || a.Build != "b-wg" || !a.PulledIn || !a.Stands() || a.Rollout != "" { + t.Errorf("the pulled-in module has no standing key of its own: %+v", a) + } + if a := get("ended"); a.Commit != "c-ended" || !a.Stands() || !a.PulledIn { + t.Errorf("an ended key is not made to stand again as pulled in: %+v", a) + } + } +} + +// The copy reads the column of the store, writes the keys, and empties the column it has copied: SentBuilds then answers the +// pulled-in module, and a second copy finds nothing to do. +func TestTheColumnOfASentMachineIsCopiedOntoItsKeysAndEmptied(t *testing.T) { + inv := ForTest(t) + ctx := t.Context() + if _, err := inv.AddNode(ctx, "laptop"); err != nil { + t.Fatal(err) + } + inv.ExecForTest(t, `update node set sent_builds = '{"mesh-wireguard": "c-wg"}'::jsonb, sent_at = now() where name = 'laptop'`) + if n, err := inv.CopySentBuilds(ctx); err != nil || n != 1 { + t.Fatalf("the copy did %d nodes (%v)", n, err) + } + got, known, err := inv.SentBuilds(ctx, "laptop") + if err != nil || !known || got["mesh-wireguard"] != "c-wg" { + t.Fatalf("SentBuilds answers %v, known %v (%v)", got, known, err) + } + if rows := inv.QueryForTest(t, `select count(*) from node where sent_builds is not null`); rows[0][0] != "0" { + t.Fatalf("the column was not emptied: %v", rows) + } + if n, err := inv.CopySentBuilds(ctx); err != nil || n != 0 { + t.Fatalf("a second copy did %d nodes (%v)", n, err) + } +} diff --git a/internal/inventory/chain_source_test.go b/internal/inventory/chain_source_test.go index 99f4c76c..d5f5cac2 100644 --- a/internal/inventory/chain_source_test.go +++ b/internal/inventory/chain_source_test.go @@ -21,10 +21,10 @@ var sqlStatement = regexp.MustCompile(`(?is)^\s*(select|insert|update|delete|alt // **No code names a table of the chain** (acceptance 12, issue 1096), found by reading the sources: a string literal of any // non-test file of internal/ or cmd/ that names one after from, into, update, join, delete from, references or table — -// build_gate and port_assignment are other words — or the column sent_builds. This is the static half of the proof; the +// build_gate and port_assignment are other words —. This is the static half of the proof; the // integration test reads the catalog. func TestNoCodeNamesATableOfTheChain(t *testing.T) { - names := regexp.MustCompile(`(?i)\b(from|into|update|join|delete\s+from|references|table)\s+(` + strings.Join(chainTables, "|") + `)\b|\bsent_builds\b`) + names := regexp.MustCompile(`(?i)\b(from|into|update|join|delete\s+from|references|table)\s+(` + strings.Join(chainTables, "|") + `)\b`) var found []string for _, root := range []string{"../../internal", "../../cmd"} { err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error { @@ -71,11 +71,10 @@ func TestNoCodeNamesATableOfTheChain(t *testing.T) { // The scan sees what it says it sees: planted, a string naming a table is found, and a comment or a word beside is not. func TestTheSourceScanFindsATablePlanted(t *testing.T) { - names := regexp.MustCompile(`(?i)\b(from|into|update|join|delete\s+from|references|table)\s+(` + strings.Join(chainTables, "|") + `)\b|\bsent_builds\b`) + names := regexp.MustCompile(`(?i)\b(from|into|update|join|delete\s+from|references|table)\s+(` + strings.Join(chainTables, "|") + `)\b`) for text, want := range map[string]bool{ "select id from build where id = $1": true, "insert into assignment (node, module)": true, - "update node set sent_builds = $1": true, "select 1 from build_gate": false, "delete from port_assignment": false, "select a from build_request a join build b": true, @@ -116,11 +115,14 @@ func TestTheLastMigrationDropsTheTablesOfTheChain(t *testing.T) { t.Errorf("migration %d does not drop %s", last.Number, table) } } - for _, want := range []string{"drop function raise_assignment_generation()", "alter table node drop column sent_builds"} { + for _, want := range []string{"drop function raise_assignment_generation()"} { if !strings.Contains(sql, want) { t.Errorf("migration %d lacks %q", last.Number, want) } } + if strings.Contains(sql, "sent_builds") { + t.Errorf("migration %d touches node.sent_builds, which stays as the source of the copy (issue 1112 drops it)", last.Number) + } if strings.Contains(sql, "build_gate") || strings.Contains(sql, "assignment_generation;") { t.Errorf("migration %d touches a table that stays", last.Number) } diff --git a/internal/inventory/migrations/0105-the-chain-leaves-the-store.sql b/internal/inventory/migrations/0105-the-chain-leaves-the-store.sql index 61c129f8..411c11b3 100644 --- a/internal/inventory/migrations/0105-the-chain-leaves-the-store.sql +++ b/internal/inventory/migrations/0105-the-chain-leaves-the-store.sql @@ -47,6 +47,6 @@ drop table pending_assignment; -- Its triggers go with it; the function they called goes here. The counter table stays: the controller raises it. drop table assignment; drop function raise_assignment_generation(); -alter table node drop column sent_builds; --- No reader of it remains. build_gate stays (issue 1097). +-- node.sent_builds STAYS, only as the source of the one-time copy onto the assignments' keys (issue 1109); dropping it is +-- issue 1112. build_gate stays (issue 1097). drop table chain_moved; diff --git a/testdata/store-tests b/testdata/store-tests index 71222538..03d58c9b 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -156,3 +156,7 @@ ./internal/inventory TestSentBuildsAnswersFromTheAssignmentsKeysOfTheNode ./internal/inventory TestASendReplacesTheWholeMapOfTheNodeAndAHandSendMakesItUnknown ./internal/inventory TestAFailedWriteOfTheLinksFailsTheRecordOfTheSend +./internal/inventory TestMigration0105RefusesAStoreWhoseRowsWereNotMovedAndDropsTheTablesOfAMovedOne +./internal/inventory TestMigration0105PassesAStoreThatNeverHeldARow +./cmd/mesh-controller TestNoTableOfTheControllersStoreHoldsBuildsRolloutsOrAssignments +./internal/inventory TestTheColumnOfASentMachineIsCopiedOntoItsKeysAndEmptied