From e0fb24126334765cac0d938f51b4c08808da2447 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 02:53:39 +0200 Subject: [PATCH] Keep a build request as the build in state asked (issue 1091) A request is the build's record under the id its outcome will carry, in state asked; the outcome takes its place and keeps the request beside it, so assign, the pending assignments and the planner read the builds bucket alone. The bucket removes no key, so a request past thirty days is not listed instead of deleted. The move folds build_request into the same records (one kind, builds, counted against both tables) and marks the requests never handed over to be asked again once: claimed before it is asked and settled after, so a crash loses the ask rather than making it twice. Also tests the chain buckets' options on a real bus, the assignments' mirror options included. --- cmd/mesh-controller/chain.go | 43 +++++ .../controller_buckets_integration_test.go | 49 +++++ internal/chain/records.go | 31 +++- internal/inventory/chain_builds.go | 44 ++++- internal/inventory/chain_move.go | 86 +++++++-- internal/inventory/chain_move_test.go | 152 ++++++++++++++++ internal/inventory/chain_requests.go | 172 ++++++++++++++++++ internal/inventory/chain_requests_test.go | 25 +++ internal/inventory/pending.go | 68 +++---- testdata/store-tests | 2 + 10 files changed, 602 insertions(+), 70 deletions(-) create mode 100644 internal/inventory/chain_requests.go create mode 100644 internal/inventory/chain_requests_test.go diff --git a/cmd/mesh-controller/chain.go b/cmd/mesh-controller/chain.go index a77986a9..e9169a01 100644 --- a/cmd/mesh-controller/chain.go +++ b/cmd/mesh-controller/chain.go @@ -52,6 +52,7 @@ func movingTheChain(ctx context.Context, inv *inventory.Inventory) { switch { case err == nil: go clearChainMove(ctx) + go askAgainSoon(ctx, inv) return case errors.As(err, &refusal): say("%v", err) @@ -122,3 +123,45 @@ func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link. movingTheChain(ctx, inv) return nil } + +// askAgain asks once more for each build request the move found never handed over (novox/hq issue 981 note 7 Q5): the +// requests a merge could not ask for, which a person's hand `build` would otherwise have to repeat. **At most once**: a +// request is claimed before it is asked and settled after, so a crash between loses the ask, which is where it stood. +func askAgain(ctx context.Context, inv *inventory.Inventory) { + pending, err := inv.RequestsToAskAgain(ctx) + if err != nil { + fmt.Printf("chain: the requests never handed over cannot be read: %v\n", err) + return + } + for _, r := range pending { + if claimed, err := inv.ClaimToAskAgain(ctx, r.ID); err != nil || !claimed { + if err != nil { + fmt.Printf("chain: %s cannot be claimed to be asked again: %v\n", r.ID, err) + } + continue + } + id, err := askABuild(ctx, buildSource{Repository: r.Repository, Seat: r.Seat}, r.Path, r.Ref) + if err != nil { + if serr := inv.AskedAgainAs(ctx, r.ID, "not asked again: "+err.Error()); serr != nil { + fmt.Printf("chain: %s was not asked again, and that could not be kept: %v\n", r.ID, serr) + } + fmt.Printf("chain: %s was never handed over and cannot be asked again: %v\n", r.ID, err) + continue + } + recordBuildRequest(ctx, inv, inventory.BuildRequest{ID: id, Repository: r.Repository, Seat: r.Seat, Path: r.Path, + Ref: r.Ref, Commit: r.Commit, For: r.For}) + if err := inv.AskedAgainAs(ctx, r.ID, id); err != nil { + fmt.Printf("chain: %s was asked again as %s, and that could not be kept: %v\n", r.ID, id, err) + } + fmt.Printf("chain: %s, never handed over, is asked again as %s\n", r.ID, id) + } +} + +// askAgainSoon waits for the controller to be serving, which an ask needs, and then asks again. +func askAgainSoon(ctx context.Context, inv *inventory.Inventory) { + select { + case <-ctx.Done(): + case <-time.After(time.Minute): + askAgain(ctx, inv) + } +} diff --git a/internal/broker/controller_buckets_integration_test.go b/internal/broker/controller_buckets_integration_test.go index 106bb3b9..9f6e8d5d 100644 --- a/internal/broker/controller_buckets_integration_test.go +++ b/internal/broker/controller_buckets_integration_test.go @@ -259,3 +259,52 @@ func TestAMergeTheOldConsumerHadNotAcknowledgedIsTheMergesConsumers(t *testing.T t.Fatalf("the merges consumer holds %v (%v); want the merge the old consumer never acknowledged", merges, err) } } + +// **The chain's buckets keep sixty-four revisions, the assignments' that were a mirror's included** (ADR 0408 §3 to §5, +// issue 1089): a bucket made as a mirror — one revision, a marker age — is brought to the chain's options without the +// server refusing the update, and a key written three times keeps three values. +func TestTheChainBucketsKeepSixtyFourRevisionsOnTheBus(t *testing.T) { + js, err := Dial(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer js.Close() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + api, err := jetstream.New(js.Conn()) + if err != nil { + t.Fatal(err) + } + // The bucket as the mirror made it. + if _, err := api.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: AssignmentsBucket, History: 1, + MaxValueSize: MirrorValueBytes, MaxBytes: MirrorBytes, Storage: jetstream.FileStorage, + LimitMarkerTTL: MirrorMarkerTTL}); err != nil { + t.Fatal(err) + } + if err := js.EnsureControllerBuckets(); err != nil { + t.Fatalf("a mirror's bucket is not brought to the chain's options: %v", err) + } + for _, name := range ChainBuckets() { + kv, err := api.KeyValue(ctx, name) + if err != nil { + t.Fatal(err) + } + status, err := kv.Status(ctx) + if err != nil { + t.Fatal(err) + } + cfg := status.(*jetstream.KeyValueBucketStatus).StreamInfo().Config + if status.History() != ChainHistory || status.TTL() != 0 || cfg.MaxMsgSize < 0 || cfg.MaxMsgSize != ChainValueBytes { + t.Errorf("%s keeps %d revisions, age %s, a value of %d bytes", name, status.History(), status.TTL(), cfg.MaxMsgSize) + } + for _, value := range []string{"one", "two", "three"} { + if _, err := kv.Put(ctx, "k", []byte(value)); err != nil { + t.Fatal(err) + } + } + history, err := kv.History(ctx, "k") + if err != nil || len(history) != 3 { + t.Errorf("%s keeps %d values of a key written three times: %v", name, len(history), err) + } + } +} diff --git a/internal/chain/records.go b/internal/chain/records.go index 01b54f9c..aa8ef078 100644 --- a/internal/chain/records.go +++ b/internal/chain/records.go @@ -53,9 +53,37 @@ type Source struct { } // The states of a build record. A record in state asked is a request that no outcome has answered (issue 1091); one -// without a state is an outcome. +// without a state is an outcome, and keeps the Request it was asked with. const BuildAsked = "asked" +// The states of a request's Reask. +const ( + ReaskPending = "pending" + ReaskClaimed = "claimed" +) + +// Request is a build as it was asked for (novox/hq issue 325, ADR 0408 §1): kept in the build's own record under its id, +// first alone as state asked and then beside the outcome that answered it. +type Request struct { + // Repository and Seat are the source as the mesh spells it; Path is the module's directory in it, Ref the branch or + // commit asked, and Commit the commit that made the ask, where one did. + Repository string `json:"repository,omitempty"` + Seat string `json:"seat,omitempty"` + Path string `json:"path,omitempty"` + Ref string `json:"ref,omitempty"` + Commit string `json:"commit,omitempty"` + // For says who asked: merge, plan, build (a person), assign. + For string `json:"for,omitempty"` + // NotAsked is why the ask could not be handed over; OutcomeUnknown why an asker that handed it over stopped waiting. + NotAsked string `json:"not-asked,omitempty"` + OutcomeUnknown string `json:"outcome-unknown,omitempty"` + At time.Time `json:"at,omitzero"` + AtTerminal bool `json:"at-terminal,omitempty"` + // Reask is, on a request the move found never handed over, "pending" until the controller asks it again, "claimed" + // while it does, and the build id it was asked as once it has: published once, never twice (issue 981 note 7 Q5). + Reask string `json:"reask,omitempty"` +} + // Build is one attempt, whichever way it went, kept under its id. It names its module, its repository and commit, its // reuse key (novox/hq ADR 0350) and what it published (ADR 0408 §1). type Build struct { @@ -79,6 +107,7 @@ type Build struct { At time.Time `json:"at,omitzero"` ReuseKey string `json:"reuse-key,omitempty"` Published *Published `json:"published,omitempty"` + Request *Request `json:"request,omitempty"` } // Commit is one repository's commit a rollout carries. diff --git a/internal/inventory/chain_builds.go b/internal/inventory/chain_builds.go index 3492d285..6a965d44 100644 --- a/internal/inventory/chain_builds.go +++ b/internal/inventory/chain_builds.go @@ -88,17 +88,46 @@ func publishedOf(made []Artifact) *chain.Published { func chainBuildKey(id string) string { return id } // chainRecordBuild keeps a build under its id. **Idempotent on the id**, as the table's insert was: a result arrives twice, -// and the first record stands. +// and the first record stands. A request nobody has answered is the record in state asked (issue 1091); the outcome +// takes its place under the same key, and keeps the request it answers. func (i *Inventory) chainRecordBuild(ctx context.Context, c *chain.Chain, b Build) error { if b.At.IsZero() { b.At = time.Now().UTC() } - if _, err := c.Builds.Create(ctx, chainBuildKey(b.ID), toChainBuild(b)); err != nil && !errors.Is(err, chain.ErrExists) { - return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err) + key := chainBuildKey(b.ID) + for try := 0; try < casTries; try++ { + next := toChainBuild(b) + cur, found, err := c.Builds.Get(ctx, key) + if err != nil { + return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err) + } + switch { + case !found: + _, err = c.Builds.Create(ctx, key, next) + if errors.Is(err, chain.ErrExists) { + continue + } + case cur.Value.State == chain.BuildAsked: + next.Request = cur.Value.Request + _, err = c.Builds.Update(ctx, key, next, cur.Revision) + if errors.Is(err, chain.ErrMoved) { + continue + } + default: + // An outcome stands: the first record is the record. + return i.RecordMirrored(ctx, b.ID, "", b.Mirrored) + } + if err != nil { + return fmt.Errorf("build %s is not kept on the bus: %w", b.ID, err) + } + return i.RecordMirrored(ctx, b.ID, "", b.Mirrored) } - return i.RecordMirrored(ctx, b.ID, "", b.Mirrored) + return fmt.Errorf("build %s is not kept on the bus: it kept moving under the write", b.ID) } +// casTries is how often a write that lost a compare-and-set reads the key again and tries once more. +const casTries = 8 + // chainBuilds is every build of the bucket, whole, in no order. func chainBuilds(ctx context.Context, c *chain.Chain) ([]Build, error) { all, err := c.Builds.All(ctx) @@ -107,6 +136,10 @@ func chainBuilds(ctx context.Context, c *chain.Chain) ([]Build, error) { } out := make([]Build, 0, len(all)) for _, k := range all { + // A request nobody has answered is no build: it has no outcome to read. + if k.Value.State == chain.BuildAsked { + continue + } out = append(out, fromChainBuild(k.Value)) } return out, nil @@ -169,7 +202,8 @@ func (i *Inventory) chainBuildList(ctx context.Context, c *chain.Chain, module s func (i *Inventory) chainBuildByID(ctx context.Context, c *chain.Chain, id string) (Build, bool, error) { k, ok, err := c.Builds.Get(ctx, chainBuildKey(id)) - if err != nil || !ok { + if err != nil || !ok || k.Value.State == chain.BuildAsked { + // A request nobody has answered has no outcome to read. return Build{}, false, err } b := fromChainBuild(k.Value) diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go index 123fa01b..bd7ee1fd 100644 --- a/internal/inventory/chain_move.go +++ b/internal/inventory/chain_move.go @@ -36,6 +36,9 @@ type MoveReport struct { // Rows is how many records the tables held, Written how many keys this move made, and Present how many already // held a value (moved by an earlier start). Rows, Written, Present int + // TableRows is how many rows of the tables those records came from: a request and its outcome are two rows and one + // record. + TableRows int // Largest is the biggest value, in bytes. Largest int // Already is a kind an earlier start moved: nothing was done. @@ -45,13 +48,13 @@ type MoveReport struct { // mover is one kind of the chain: its rows, read from its tables, as the keys and values they become. type mover struct { kind string - // rows reads the tables: each record's key and its value, whole. - rows func(ctx context.Context) ([]moveRow, error) + // rows reads the tables: each record's key and its value, whole, and how many rows of the tables they came from. + rows func(ctx context.Context) ([]moveRow, int, error) // create writes one record under its key; chain.ErrExists when the key holds a value. create func(ctx context.Context, c *chain.Chain, row moveRow) error // keys are the keys the bucket holds. keys func(ctx context.Context, c *chain.Chain) ([]string, error) - // count is how many rows the tables hold now. + // count is how many rows the tables hold now, counted as rows reports them. 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 @@ -66,21 +69,46 @@ type moveRow struct { func (i *Inventory) movers() []mover { return []mover{{ kind: KindBuilds, - rows: func(ctx context.Context) ([]moveRow, error) { + rows: func(ctx context.Context) ([]moveRow, int, error) { builds, err := i.allBuildRows(ctx) if err != nil { - return nil, err + return nil, 0, err } - out := make([]moveRow, 0, len(builds)) + requests, err := i.allRequestRows(ctx) + if err != nil { + return nil, 0, err + } + // A build request is the build's record in state asked, and the request stays beside the outcome that + // answered it: one record under one id (issue 1091). + byID := make(map[string]*chain.Build, len(builds)) + var order []string for _, b := range builds { v := toChainBuild(b) - body, err := json.Marshal(v) - if err != nil { - return nil, err - } - out = append(out, moveRow{key: chainBuildKey(b.ID), value: v, size: len(body)}) + byID[b.ID] = &v + order = append(order, b.ID) } - return out, nil + for _, r := range requests { + if v, answered := byID[r.ID]; answered { + v.Request = toRequest(r) + continue + } + v := askedRecord(r) + if r.NotAsked != "" { + // Never handed over: asked again once, after the move (askAgain). + v.Request.Reask = chain.ReaskPending + } + byID[r.ID] = &v + order = append(order, r.ID) + } + out := make([]moveRow, 0, len(order)) + for _, id := range order { + body, err := json.Marshal(byID[id]) + if err != nil { + return nil, 0, err + } + out = append(out, moveRow{key: chainBuildKey(id), value: *byID[id], size: len(body)}) + } + return out, len(builds) + len(requests), nil }, create: func(ctx context.Context, c *chain.Chain, row moveRow) error { _, err := c.Builds.Create(ctx, row.key, row.value.(chain.Build)) @@ -89,10 +117,10 @@ func (i *Inventory) movers() []mover { keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Builds.Keys(ctx) }, count: func(ctx context.Context, tx pgx.Tx) (int, error) { var n int - err := tx.QueryRow(ctx, `select count(*) from build`).Scan(&n) + err := tx.QueryRow(ctx, `select (select count(*) from build) + (select count(*) from build_request)`).Scan(&n) return n, err }, - lock: `lock table build in share mode`, + lock: `lock table build, build_request in share mode`, }} } @@ -131,11 +159,11 @@ func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...a return report, nil } for pass := 1; pass <= movePasses; pass++ { - rows, err := m.rows(ctx) + rows, tableRows, err := m.rows(ctx) if err != nil { return report, fmt.Errorf("the %s cannot be read from the tables: %w", m.kind, err) } - report.Rows, report.Largest = len(rows), 0 + report.Rows, report.TableRows, report.Largest = len(rows), tableRows, 0 // Every value measured before any is written: a value over the bus's largest stops the move with nothing // written (ADR 0408 §6), never cut. for _, r := range rows { @@ -152,6 +180,8 @@ func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...a report.Written++ case errors.Is(err, chain.ErrExists): report.Present++ + case errors.Is(err, chain.ErrBadKey): + return report, &Refusal{m.kind, fmt.Sprintf("%s cannot be a key of the bus: %v", r.key, err)} default: return report, fmt.Errorf("%s %s is not written to the bus: %w", m.kind, r.key, err) } @@ -199,7 +229,7 @@ func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool if err != nil { return false, err } - if n != r.Rows { + if n != r.TableRows { return false, nil } counts, _ := json.Marshal(map[string]int{m.kind: r.Rows}) @@ -263,3 +293,25 @@ func (i *Inventory) allBuildRows(ctx context.Context) ([]Build, error) { } return out, rows.Err() } + +// allRequestRows is every build request of the table. +func (i *Inventory) allRequestRows(ctx context.Context) ([]BuildRequest, error) { + rows, err := i.store.Pool().Query(ctx, + `select id, repository, seat, source_path, ref, commit_hash, asked_for, coalesce(not_asked, ''), + coalesce(outcome_unknown, ''), asked_at, at_terminal + from build_request order by asked_at, id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []BuildRequest + for rows.Next() { + var a BuildRequest + if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked, + &a.OutcomeUnknown, &a.At, &a.AtTerminal); err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go index 2a4cc219..6cd193cb 100644 --- a/internal/inventory/chain_move_test.go +++ b/internal/inventory/chain_move_test.go @@ -6,7 +6,9 @@ import ( "context" "errors" "reflect" + "slices" "testing" + "time" "github.com/novox/mesh-controller/internal/chain" ) @@ -270,3 +272,153 @@ func TestBuildsOnTheBusAreReadBackByEveryReader(t *testing.T) { t.Fatalf("fingerprints %v: %v", prints, err) } } + +// A build request row as the store held it before the move. +func aRequestRow(t *testing.T, inv *Inventory, id, notAsked string) { + t.Helper() + inv.ExecForTest(t, + `insert into build_request (id, repository, seat, source_path, ref, commit_hash, asked_for, not_asked, asked_at) + values ($1::text, 'novox/notes', '', 'modules/notes', 'main', 'c1', 'merge', nullif($2::text, ''), now() - interval '1 hour')`, + id, notAsked) +} + +// **A request is the build record in state asked, and its outcome takes its place and keeps it** (ADR 0408 §1, issue +// 1091): no outcome to read until it comes, the marks of a request that nobody answered kept in the record, and none +// after the outcome is heard. The table of requests takes none. +func TestARequestIsTheBuildInStateAskedAndItsOutcomeTakesItsPlace(t *testing.T) { + inv := fresh(t) + ctx := context.Background() + ask := BuildRequest{ID: "build-1", Repository: "novox/notes", Path: "modules/notes", Ref: "main", For: "build", AtTerminal: true} + if err := inv.RecordBuildRequest(ctx, ask); err != nil { + t.Fatal(err) + } + if err := inv.RecordBuildRequest(ctx, ask); err != nil { + t.Fatalf("a request kept twice: %v", err) + } + if n := rowsInTheTable(t, inv, "build_request"); n != "0" { + t.Fatalf("the table took %s requests", n) + } + if builds, err := inv.Builds(ctx, "", 10); err != nil || len(builds) != 0 { + t.Fatalf("a request nobody answered is a build: %+v %v", builds, err) + } + if _, found, _ := inv.BuildByID(ctx, "build-1"); found { + t.Fatal("a request nobody answered has an outcome") + } + got, found, err := inv.BuildRequestByID(ctx, "build-1") + if err != nil || !found || got.Repository != "novox/notes" || got.For != "build" || !got.AtTerminal || got.At.IsZero() { + t.Fatalf("the request reads back as %+v %v %v", got, found, err) + } + if err := inv.MarkOutcomeUnknown(ctx, "build-1", "stopped waiting"); err != nil { + t.Fatal(err) + } + asked, err := inv.RequestsNamed(ctx, "notes") + if err != nil || len(asked) != 1 || asked[0].Heard || asked[0].OutcomeUnknown != "stopped waiting" { + t.Fatalf("the request is %+v: %v", asked, err) + } + if names, err := inv.RequestedNames(ctx); err != nil || !reflect.DeepEqual(names, []string{"notes"}) { + t.Fatalf("requested names %v: %v", names, err) + } + + // The outcome comes: the same key, now a build, with the request beside it. + built := aBuild("build-1", "notes", "") + if err := inv.RecordBuild(ctx, built); err != nil { + t.Fatal(err) + } + if one, found, err := inv.BuildByID(ctx, "build-1"); err != nil || !found || one.Module != "notes" { + t.Fatalf("the outcome reads as %+v %v %v", one, found, err) + } + answered, err := inv.RequestsNamed(ctx, "notes") + if err != nil || len(answered) != 1 || !answered[0].Heard || answered[0].Module != "notes" || answered[0].Failed != "" { + t.Fatalf("the request answered is %+v: %v", answered, err) + } + if again, _, _ := inv.BuildRequestByID(ctx, "build-1"); again.For != "build" || again.OutcomeUnknown != "stopped waiting" { + t.Fatalf("the request was lost with the outcome: %+v", again) + } + // Heard: the marks of an unanswered request change nothing. + if err := inv.MarkNotAsked(ctx, "build-1", "late"); err != nil { + t.Fatal(err) + } + if again, _, _ := inv.BuildRequestByID(ctx, "build-1"); again.NotAsked != "" { + t.Fatalf("an answered request was marked not asked: %+v", again) + } + if history, _ := inv.chain.Builds.History(ctx, "build-1"); len(history) != 3 { + // asked, marked unknown, answered. + t.Fatalf("the record has %d values", len(history)) + } + // An outcome heard before its request is kept, and the request lands beside it. + if err := inv.RecordBuild(ctx, aBuild("build-2", "wiki", "")); err != nil { + t.Fatal(err) + } + if err := inv.RecordBuildRequest(ctx, BuildRequest{ID: "build-2", Repository: "novox/wiki", Path: "modules/wiki", For: "plan"}); err != nil { + t.Fatal(err) + } + if late, found, _ := inv.BuildRequestByID(ctx, "build-2"); !found || late.For != "plan" { + t.Fatalf("the request kept after its outcome reads %+v", late) + } + // A request older than the time they are kept is no longer listed. + if err := inv.RecordBuildRequest(ctx, BuildRequest{ID: "build-old", Repository: "novox/old", Path: "modules/old", + At: time.Now().Add(-2 * KeptFor)}); err != nil { + t.Fatal(err) + } + if names, _ := inv.RequestedNames(ctx); slices.Contains(names, "old") { + t.Fatalf("a request past its time is listed: %v", names) + } +} + +// **The move carries the build requests**: an answered one beside its outcome, an unanswered one as state asked, and a +// request never handed over as state asked marked to be asked again once (issue 981 note 7 Q5), which is claimed before it +// is asked and so asked at most once. +func TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + aRequestRow(t, inv, "build-1", "") + aRequestRow(t, inv, "build-2", "") + aRequestRow(t, inv, "not-asked-c1-modules-notes", "no seat holder") + + if requests, err := inv.RequestsNamed(ctx, "notes"); err != nil || len(requests) != 3 { + t.Fatalf("before the move the table holds %d requests: %v", len(requests), err) + } + reports, err := inv.MoveChain(ctx, nil) + if err != nil { + t.Fatal(err) + } + if reports[0].Rows != 3 || reports[0].TableRows != 4 { + t.Fatalf("the move reported %+v", reports[0]) + } + 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" { + t.Fatalf("an answered request moved as %+v", one.Value) + } + two, _, _ := inv.chain.Builds.Get(ctx, "build-2") + if two.Value.State != chain.BuildAsked || two.Value.Request.Reask != "" { + t.Fatalf("an unanswered request moved as %+v", two.Value) + } + never, _, _ := inv.chain.Builds.Get(ctx, "not-asked-c1-modules-notes") + if never.Value.State != chain.BuildAsked || never.Value.Request.NotAsked != "no seat holder" || never.Value.Request.Reask != chain.ReaskPending { + t.Fatalf("a request never handed over moved as %+v", never.Value) + } + requests, err := inv.RequestsNamed(ctx, "notes") + if err != nil || len(requests) != 3 { + t.Fatalf("after the move the bucket holds %d requests: %v", len(requests), err) + } + toAsk, err := inv.RequestsToAskAgain(ctx) + if err != nil || len(toAsk) != 1 || toAsk[0].ID != "not-asked-c1-modules-notes" { + t.Fatalf("to ask again: %+v %v", toAsk, err) + } + first, err := inv.ClaimToAskAgain(ctx, toAsk[0].ID) + second, err2 := inv.ClaimToAskAgain(ctx, toAsk[0].ID) + if err != nil || err2 != nil || !first || second { + t.Fatalf("claimed %v then %v: %v %v", first, second, err, err2) + } + if left, _ := inv.RequestsToAskAgain(ctx); len(left) != 0 { + t.Fatalf("a claimed request is still to be asked: %+v", left) + } + if err := inv.AskedAgainAs(ctx, toAsk[0].ID, "build-9"); err != nil { + t.Fatal(err) + } + if done, _, _ := inv.BuildRequestByID(ctx, toAsk[0].ID); done.NotAsked != "no seat holder" { + t.Fatalf("the request lost its words: %+v", done) + } +} diff --git a/internal/inventory/chain_requests.go b/internal/inventory/chain_requests.go new file mode 100644 index 00000000..230ba9c3 --- /dev/null +++ b/internal/inventory/chain_requests.go @@ -0,0 +1,172 @@ +package inventory + +import ( + "context" + "errors" + "fmt" + "sort" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// Build requests on the bus (novox/hq ADR 0408, issue 1091): a request is the build record in state asked, under the id its +// outcome will carry; the outcome takes its place and keeps the request. The bucket removes no key, so a request is not +// deleted after KeptFor: it is no longer listed. + +func toRequest(a BuildRequest) *chain.Request { + return &chain.Request{Repository: a.Repository, Seat: a.Seat, Path: strings.Trim(a.Path, "/"), Ref: a.Ref, Commit: a.Commit, + For: a.For, NotAsked: a.NotAsked, OutcomeUnknown: a.OutcomeUnknown, At: a.At.UTC(), AtTerminal: a.AtTerminal} +} + +func fromRequest(id string, r chain.Request) BuildRequest { + return BuildRequest{ID: id, Repository: r.Repository, Seat: r.Seat, Path: r.Path, Ref: r.Ref, Commit: r.Commit, For: r.For, + NotAsked: r.NotAsked, OutcomeUnknown: r.OutcomeUnknown, At: r.At, AtTerminal: r.AtTerminal} +} + +// askedRecord is a request as the build record in state asked. +func askedRecord(a BuildRequest) chain.Build { + r := toRequest(a) + return chain.Build{ID: a.ID, State: chain.BuildAsked, Repository: a.Repository, Ref: a.Ref, Path: r.Path, Asked: r.At, Request: r} +} + +func (i *Inventory) chainRecordBuildRequest(ctx context.Context, c *chain.Chain, a BuildRequest) error { + if a.At.IsZero() { + a.At = time.Now().UTC() + } + key := chainBuildKey(a.ID) + for try := 0; try < casTries; try++ { + _, err := c.Builds.Create(ctx, key, askedRecord(a)) + if err == nil { + return nil + } + if !errors.Is(err, chain.ErrExists) { + return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err) + } + // The key holds a value: the request is kept already, or its outcome came first and has no request beside it. + cur, found, err := c.Builds.Get(ctx, key) + if err != nil { + return err + } + if !found || cur.Value.Request != nil { + if !found { + continue + } + return nil + } + cur.Value.Request = toRequest(a) + if _, err := c.Builds.Update(ctx, key, cur.Value, cur.Revision); err == nil { + return nil + } else if !errors.Is(err, chain.ErrMoved) { + return fmt.Errorf("build request %s is not kept on the bus: %w", a.ID, err) + } + } + return fmt.Errorf("build request %s is not kept on the bus: it kept moving under the write", a.ID) +} + +func (i *Inventory) chainBuildRequestByID(ctx context.Context, c *chain.Chain, id string) (BuildRequest, bool, error) { + k, found, err := c.Builds.Get(ctx, chainBuildKey(id)) + if err != nil || !found || k.Value.Request == nil { + return BuildRequest{}, false, err + } + return fromRequest(id, *k.Value.Request), true, nil +} + +// chainMarkRequest changes a request that has no outcome yet, by compare-and-set; one answered, or none kept, is left. +func (i *Inventory) chainMarkRequest(ctx context.Context, c *chain.Chain, id string, mark func(*chain.Request)) error { + key := chainBuildKey(id) + for try := 0; try < casTries; try++ { + cur, found, err := c.Builds.Get(ctx, key) + if err != nil || !found || cur.Value.State != chain.BuildAsked || cur.Value.Request == nil { + return err + } + mark(cur.Value.Request) + if _, err := c.Builds.Update(ctx, key, cur.Value, cur.Revision); err == nil { + return nil + } else if !errors.Is(err, chain.ErrMoved) { + return fmt.Errorf("build request %s is not marked on the bus: %w", id, err) + } + } + return fmt.Errorf("build request %s is not marked: it kept moving under the write", id) +} + +// chainRequests is every request kept, newest first, each with its outcome where its build was heard. Requests older than +// KeptFor are not listed: the bucket keeps them, and no reader needs them. +func (i *Inventory) chainRequests(ctx context.Context, c *chain.Chain) ([]RequestOutcome, error) { + all, err := c.Builds.All(ctx) + if err != nil { + return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) + } + oldest := time.Now().Add(-KeptFor) + var out []RequestOutcome + for id, k := range all { + r := k.Value.Request + if r == nil || !r.At.After(oldest) { + continue + } + a := RequestOutcome{BuildRequest: fromRequest(id, *r), HeardAt: r.At} + if k.Value.State != chain.BuildAsked { + a.Heard, a.Failed, a.Module, a.HeardAt = true, k.Value.Failed, k.Value.Module, k.Value.At + } + out = append(out, a) + } + sort.Slice(out, func(a, b int) bool { + if !out[a].At.Equal(out[b].At) { + return out[a].At.After(out[b].At) + } + return out[a].ID > out[b].ID + }) + return out, nil +} + +// RequestsToAskAgain are the requests the move found never handed over (issue 981 note 7 Q5), not yet asked again: each is +// asked once, and never twice. +func (i *Inventory) RequestsToAskAgain(ctx context.Context) ([]BuildRequest, error) { + c, on, err := i.ChainReads(ctx, KindBuilds) + if err != nil || !on { + return nil, err + } + all, err := c.Builds.All(ctx) + if err != nil { + return nil, err + } + var out []BuildRequest + for id, k := range all { + if k.Value.State == chain.BuildAsked && k.Value.Request != nil && k.Value.Request.Reask == chain.ReaskPending { + out = append(out, fromRequest(id, *k.Value.Request)) + } + } + sort.Slice(out, func(a, b int) bool { return out[a].ID < out[b].ID }) + return out, nil +} + +// ClaimToAskAgain takes a request to ask again, by compare-and-set: true for the one process that took it. **Claimed +// before it is asked and settled after**, so a crash between loses the ask rather than making it twice. +func (i *Inventory) ClaimToAskAgain(ctx context.Context, id string) (bool, error) { + c, on, err := i.ChainReads(ctx, KindBuilds) + if err != nil || !on { + return false, err + } + cur, found, err := c.Builds.Get(ctx, chainBuildKey(id)) + if err != nil || !found || cur.Value.Request == nil || cur.Value.Request.Reask != chain.ReaskPending { + return false, err + } + cur.Value.Request.Reask = chain.ReaskClaimed + if _, err := c.Builds.Update(ctx, chainBuildKey(id), cur.Value, cur.Revision); err != nil { + if errors.Is(err, chain.ErrMoved) { + return false, nil + } + return false, err + } + return true, nil +} + +// AskedAgainAs records the build id a claimed request was asked as. +func (i *Inventory) AskedAgainAs(ctx context.Context, id, asked string) error { + c, on, err := i.ChainReads(ctx, KindBuilds) + if err != nil || !on { + return err + } + return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.Reask = asked }) +} diff --git a/internal/inventory/chain_requests_test.go b/internal/inventory/chain_requests_test.go new file mode 100644 index 00000000..6f99b143 --- /dev/null +++ b/internal/inventory/chain_requests_test.go @@ -0,0 +1,25 @@ +package inventory + +import ( + "reflect" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// **A request is the build record in state asked, under the id its outcome will carry** (ADR 0408 §1, issue 1091): the +// whole ask is in it, with a path trimmed as the table kept it. +func TestABuildRequestIsTheBuildRecordInStateAsked(t *testing.T) { + at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + ask := BuildRequest{ID: "build-1", Repository: "https://forge.invalid/notes.git", Seat: "", Path: "/modules/notes/", Ref: "main", + Commit: "c1", For: "merge", NotAsked: "", OutcomeUnknown: "", At: at, AtTerminal: true} + record := askedRecord(ask) + if record.State != chain.BuildAsked || record.ID != "build-1" || record.Path != "modules/notes" || !record.Asked.Equal(at) { + t.Fatalf("the request is %+v", record) + } + if got := fromRequest(record.ID, *record.Request); !reflect.DeepEqual(got, BuildRequest{ID: "build-1", + Repository: ask.Repository, Path: "modules/notes", Ref: "main", Commit: "c1", For: "merge", At: at, AtTerminal: true}) { + t.Fatalf("the request reads back as %+v", got) + } +} diff --git a/internal/inventory/pending.go b/internal/inventory/pending.go index 8b739e28..7f8258d5 100644 --- a/internal/inventory/pending.go +++ b/internal/inventory/pending.go @@ -3,7 +3,6 @@ package inventory import ( "context" "errors" - "fmt" "path" "strings" "time" @@ -70,6 +69,11 @@ type RequestOutcome struct { // RecordBuildRequest keeps one build request, once it is asked. Idempotent on the id. Requests older than // KeptFor are deleted in the same act. func (i *Inventory) RecordBuildRequest(ctx context.Context, a BuildRequest) error { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return err + } else if on { + return i.chainRecordBuildRequest(ctx, c, a) + } at := a.At if at.IsZero() { at = time.Now().UTC() @@ -93,6 +97,11 @@ func (i *Inventory) RecordBuildRequest(ctx context.Context, a BuildRequest) erro // BuildRequestByID is the build request kept under this id, and whether one was kept. func (i *Inventory) BuildRequestByID(ctx context.Context, id string) (BuildRequest, bool, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return BuildRequest{}, false, err + } else if on { + return i.chainBuildRequestByID(ctx, c, id) + } var a BuildRequest err := i.store.Pool().QueryRow(ctx, `select id, repository, seat, source_path, ref, commit_hash, asked_for, at_terminal, asked_at @@ -107,6 +116,11 @@ func (i *Inventory) BuildRequestByID(ctx context.Context, id string) (BuildReque // MarkNotAsked says a kept build request was never handed over: the words are kept, unless its outcome was // heard first. func (i *Inventory) MarkNotAsked(ctx context.Context, id, why string) error { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return err + } else if on { + return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.NotAsked = why }) + } if heard, err := i.outcomeHeard(ctx, id); err != nil || heard { return err } @@ -117,6 +131,11 @@ func (i *Inventory) MarkNotAsked(ctx context.Context, id, why string) error { // MarkOutcomeUnknown says the asker of a build it handed over stopped waiting for it: asked, outcome unknown. // The build may still run; its outcome, when heard, is the last word. func (i *Inventory) MarkOutcomeUnknown(ctx context.Context, id, why string) error { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return err + } else if on { + return i.chainMarkRequest(ctx, c, id, func(r *chain.Request) { r.OutcomeUnknown = why }) + } if heard, err := i.outcomeHeard(ctx, id); err != nil || heard { return err } @@ -124,14 +143,8 @@ func (i *Inventory) MarkOutcomeUnknown(ctx context.Context, id, why string) erro return err } -// outcomeHeard says a build of this id was recorded: its outcome was taken in. +// outcomeHeard says a build of this id was recorded in the table: its outcome was taken in. func (i *Inventory) outcomeHeard(ctx context.Context, id string) (bool, error) { - if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { - return false, err - } else if on { - _, heard, err := c.Builds.Get(ctx, chainBuildKey(id)) - return heard, err - } var heard bool err := i.store.Pool().QueryRow(ctx, `select exists (select 1 from build where id = $1)`, id).Scan(&heard) return heard, err @@ -199,45 +212,6 @@ func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) { return out, rows.Err() } -// chainRequests is requests with each outcome read from the builds bucket, where the table's join read it. -func (i *Inventory) chainRequests(ctx context.Context, c *chain.Chain) ([]RequestOutcome, error) { - rows, err := i.store.Pool().Query(ctx, - `select id, repository, seat, source_path, ref, commit_hash, asked_for, - coalesce(not_asked, ''), coalesce(outcome_unknown, ''), asked_at - from build_request - order by asked_at desc, id desc`) - if err != nil { - return nil, err - } - defer rows.Close() - var out []RequestOutcome - for rows.Next() { - var a RequestOutcome - if err := rows.Scan(&a.ID, &a.Repository, &a.Seat, &a.Path, &a.Ref, &a.Commit, &a.For, &a.NotAsked, - &a.OutcomeUnknown, &a.At); err != nil { - return nil, err - } - a.HeardAt = a.At - out = append(out, a) - } - if err := rows.Err(); err != nil { - return nil, err - } - if len(out) == 0 { - return out, nil - } - built, err := c.Builds.All(ctx) - if err != nil { - return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) - } - for k := range out { - if b, heard := built[chainBuildKey(out[k].ID)]; heard { - out[k].Heard, out[k].Failed, out[k].Module, out[k].HeardAt = true, b.Value.Failed, b.Value.Module, b.Value.At - } - } - return out, nil -} - // The states of a pending assignment. const ( PendingWaiting = "waiting" diff --git a/testdata/store-tests b/testdata/store-tests index c3536357..52dcabd7 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -145,3 +145,5 @@ ./internal/inventory TestTheMoveRefusesOnACountThatDiffers ./internal/inventory TestTheMoveWithoutTheLeaseWritesNothing ./internal/inventory TestBuildsOnTheBusAreReadBackByEveryReader +./internal/inventory TestARequestIsTheBuildInStateAskedAndItsOutcomeTakesItsPlace +./internal/inventory TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain