From f89d8703082536f876b2cd52fc3932201927c9ac Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 02:49:51 +0200 Subject: [PATCH] Keep builds in the builds bucket and move the build rows there at start (issue 1090) Once chain_moved records the kind, RecordBuild and every build read go through the chain (reuse key and published added); before that the tables are read as before. The serving controller copies the rows at its start, holding the lease: idempotent per key, rows counted against keys, the largest value measured, a refusal (urgent condition, nothing switched) on a count that differs or a value over the bus's largest, and a re-read when the table took a row meanwhile. ForTest attaches the chain in memory with builds moved; ForTestBefore leaves it unmoved. Build requests, gates and the other kinds are still in their tables. --- cmd/mesh-controller/chain.go | 124 ++++++ cmd/mesh-controller/push.go | 5 + cmd/mesh-controller/stores.go | 3 + internal/chain/bus.go | 4 +- internal/chain/chain.go | 69 ++- internal/chain/chain_test.go | 32 +- internal/inventory/builds.go | 39 ++ internal/inventory/builds_test.go | 11 +- internal/inventory/chain.go | 113 +++++ internal/inventory/chain_builds.go | 536 ++++++++++++++++++++++++ internal/inventory/chain_builds_test.go | 100 +++++ internal/inventory/chain_move.go | 265 ++++++++++++ internal/inventory/chain_move_test.go | 272 ++++++++++++ internal/inventory/collection.go | 133 ++++-- internal/inventory/fortest.go | 10 + internal/inventory/gate.go | 71 +++- internal/inventory/nodes.go | 8 +- internal/inventory/pending.go | 74 +++- internal/inventory/source.go | 5 + testdata/store-tests | 6 + 20 files changed, 1790 insertions(+), 90 deletions(-) create mode 100644 cmd/mesh-controller/chain.go create mode 100644 internal/inventory/chain.go create mode 100644 internal/inventory/chain_builds.go create mode 100644 internal/inventory/chain_builds_test.go create mode 100644 internal/inventory/chain_move.go create mode 100644 internal/inventory/chain_move_test.go diff --git a/cmd/mesh-controller/chain.go b/cmd/mesh-controller/chain.go new file mode 100644 index 00000000..a77986a9 --- /dev/null +++ b/cmd/mesh-controller/chain.go @@ -0,0 +1,124 @@ +package main + +import ( + "context" + "errors" + "fmt" + "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" +) + +// The chain on the bus (novox/hq ADR 0408, design 47 milestone 5): builds, rollouts and assignments are keys of the +// bus's key-value store, written by the controller that holds the lease. The tables of the controller's store are +// copied into them once, by the serving controller at its start, and a kind is read from its bucket only once the move +// has recorded it (inventory.ChainReads). + +// chainHolds is whether this process may write the chain now: the lease's gate, the one every act passes. +func chainHolds(ctx context.Context) error { + _, err := theLease.epoch(ctx) + return err +} + +// openTheChain opens the chain's buckets on the bus: the serving controller's connection, or one dialled for a command. +func openTheChain(ctx context.Context) (*chain.Chain, error) { + js, err := aBus() + if err != nil { + return nil, err + } + // Not closed: a borrowed connection closes nothing, and a dialled one lasts as long as the command. + return chain.OnTheBus(ctx, js.Conn(), chainHolds) +} + +// kindChainMoveRefused is the condition of a move that stopped without switching a kind. +const kindChainMoveRefused = "chain-move-refused" + +// chainMoveRetry is how long a move that failed for want of the bus waits before it tries again. +const chainMoveRetry = 30 * time.Second + +// movingTheChain copies the tables into the buckets and records it, at the serving controller's start: after the buckets +// exist and the lease is held, before anything reads a kind that has not moved. **The first try is made here, before the +// controller serves**, so what it serves is read from one place. A move that fails for want of the bus is tried again +// every half minute while the controller serves from the tables; one that is refused — a count that differs, a value +// over the bus's largest — is raised as an urgent condition, and the tables stay what is read: nothing is half-switched. +func movingTheChain(ctx context.Context, inv *inventory.Inventory) { + say := func(format string, args ...any) { fmt.Printf("chain: "+format+"\n", args...) } + for { + _, err := inv.MoveChain(ctx, say) + var refusal *inventory.Refusal + switch { + case err == nil: + go clearChainMove(ctx) + return + case errors.As(err, &refusal): + say("%v", err) + go raiseChainMove(ctx, refusal) + return + case ctx.Err() != nil: + return + default: + say("the move to the bus did not finish, and the tables are read until it does: %v", err) + } + select { + case <-ctx.Done(): + return + case <-time.After(chainMoveRetry): + } + } +} + +// chainMoveObservation is the urgent condition for a refused move. +func chainMoveObservation(r *inventory.Refusal) conditions.Observation { + return conditions.Observation{Scope: conditions.ScopeBus, ID: "chain." + r.Kind, Token: kindChainMoveRefused, + Kind: kindChainMoveRefused, Severity: conditions.Urgent, Resolver: conditions.ResolverOperator, + Summary: "the controller's " + r.Kind + " were not moved to the bus: " + r.Why, Source: "chain move"} +} + +// whenKept runs f once the serving controller keeps conditions: the move runs before they are kept. +func whenKept(ctx context.Context, f func(k *conditions.Keeper)) { + for conditionsFrom == nil { + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } + } + f(conditionsFrom) +} + +func raiseChainMove(ctx context.Context, r *inventory.Refusal) { + whenKept(ctx, func(k *conditions.Keeper) { + if _, err := k.Observe(ctx, chainMoveObservation(r)); err != nil { + fmt.Printf("chain: the condition for the refused move could not be raised: %v\n", err) + } + }) +} + +func clearChainMove(ctx context.Context) { + whenKept(ctx, func(k *conditions.Keeper) { + for _, kind := range []string{inventory.KindBuilds} { + key := chainMoveObservation(&inventory.Refusal{Kind: kind}).Key() + if _, err := k.Clear(ctx, key, "the "+kind+" are on the bus"); err != nil { + fmt.Printf("chain: the condition for the move of the %s could not be cleared: %v\n", kind, err) + } + } + }) +} + +// attachTheChain gives the inventory of the serving controller the chain on its connection, then moves what has not moved. +func attachTheChain(ctx context.Context, inv *inventory.Inventory, server *link.Server) error { + bus, ok := server.Bus().(link.OverNATS) + if !ok { + return errors.New("the chain lives on the bus, and this control plane is not on it") + } + c, err := chain.OnTheBus(ctx, bus.Conn, chainHolds) + if err != nil { + return err + } + inv.UseChain(c) + movingTheChain(ctx, inv) + return nil +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 3008cc42..2ae089e6 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -172,6 +172,11 @@ func serve(ctx context.Context) (err error) { return err } defer server.Close() + // The chain on the bus, and the move of what the tables still hold into it (novox/hq ADR 0408 §8): before anything + // reads builds, rollouts or assignments. + if err := attachTheChain(ctx, inv, server); err != nil { + return err + } // The bus's own objects, asserted on every start. **Not created once at genesis**: a stream // somebody deleted, a mesh raised from a restored backup, or a bus whose data directory was diff --git a/cmd/mesh-controller/stores.go b/cmd/mesh-controller/stores.go index 5fef54d7..ff727dbc 100644 --- a/cmd/mesh-controller/stores.go +++ b/cmd/mesh-controller/stores.go @@ -132,6 +132,9 @@ func openInventory(ctx context.Context) (*inventory.Inventory, error) { } // And where config values are read once the move of config has run (issue 664): mesh-config_resolved. inv.ReadsConfigFrom(readConfigValues) + // And the chain, opened when a kind that moved to the bus is first read (novox/hq ADR 0408): a command that reads none + // has no use of a connection. + inv.UseChainOpener(openTheChain) return inv, nil } diff --git a/internal/chain/bus.go b/internal/chain/bus.go index 97c5b872..f34670b2 100644 --- a/internal/chain/bus.go +++ b/internal/chain/bus.go @@ -13,7 +13,7 @@ import ( // OnTheBus opens the chain's three buckets on a connection. The controller asserts them at its start // (broker.EnsureControllerBuckets), so one older than this has not, and the error says so. -func OnTheBus(ctx context.Context, conn *nats.Conn, holds func() bool) (*Chain, error) { +func OnTheBus(ctx context.Context, conn *nats.Conn, holds Holds) (*Chain, error) { api, err := jetstream.New(conn) if err != nil { return nil, err @@ -85,7 +85,7 @@ func (b busBucket) Keys(ctx context.Context) ([]string, error) { } func (b busBucket) Watch(ctx context.Context) (<-chan Entry, error) { - w, err := b.kv.WatchAll(ctx) + w, err := b.kv.WatchAll(ctx, jetstream.IgnoreDeletes()) if err != nil { return nil, err } diff --git a/internal/chain/chain.go b/internal/chain/chain.go index d0ec79a7..0b987229 100644 --- a/internal/chain/chain.go +++ b/internal/chain/chain.go @@ -3,7 +3,10 @@ package chain import ( "context" "encoding/json" + "errors" "fmt" + "regexp" + "strings" "github.com/novox/mesh-controller/internal/broker" ) @@ -12,9 +15,13 @@ import ( // controller holds the lease** and writes nothing when it does not (ADR 0408 §2). type Bucket[T any] struct { b Backend - holds func() bool + holds Holds } +// Holds says whether this controller may write now: nil when it holds the lease, and why not when it does not. The +// serving controller answers from its lease, and a command from the holder's (the epoch gate). +type Holds func(ctx context.Context) error + // Kept is a record as the bucket holds it: with its revision, the one a write must find. type Kept[T any] struct { Key string @@ -22,10 +29,22 @@ type Kept[T any] struct { Revision uint64 } -func (s Bucket[T]) guard(value []byte) error { - if s.holds == nil || !s.holds() { +// validKey is what the bus takes as a key: letters, digits and - _ / = . with no dot at either end. +var validKey = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`) + +// ErrBadKey is a key the bus does not take: refused loudly, never rewritten into another. +var ErrBadKey = errors.New("the bus takes no key of that spelling") + +func (s Bucket[T]) guard(ctx context.Context, key string, value []byte) error { + if s.holds == nil { return ErrNotHeld } + if err := s.holds(ctx); err != nil { + return fmt.Errorf("%w: %v", ErrNotHeld, err) + } + if !validKey.MatchString(key) || strings.HasPrefix(key, ".") || strings.HasSuffix(key, ".") { + return fmt.Errorf("%w: %q", ErrBadKey, key) + } if len(value) > broker.ChainValueBytes { return fmt.Errorf("%w: %d bytes, the most is %d", ErrTooLarge, len(value), broker.ChainValueBytes) } @@ -57,7 +76,7 @@ func (s Bucket[T]) Create(ctx context.Context, key string, v T) (uint64, error) if err != nil { return 0, err } - if err := s.guard(body); err != nil { + if err := s.guard(ctx, key, body); err != nil { return 0, err } return s.b.Create(ctx, key, body) @@ -69,29 +88,39 @@ func (s Bucket[T]) Update(ctx context.Context, key string, v T, revision uint64) if err != nil { return 0, err } - if err := s.guard(body); err != nil { + if err := s.guard(ctx, key, body); err != nil { return 0, err } return s.b.Update(ctx, key, body, revision) } -// All is every record, by key. An error is an error: never an empty bucket. +// All is every record, by key, read in one pass of a watch. An error is an error: never an empty bucket. func (s Bucket[T]) All(ctx context.Context) (map[string]Kept[T], error) { - keys, err := s.b.Keys(ctx) + ctx, stop := context.WithCancel(ctx) + defer stop() + w, err := s.b.Watch(ctx) if err != nil { return nil, err } - out := make(map[string]Kept[T], len(keys)) - for _, key := range keys { - k, ok, err := s.Get(ctx, key) - if err != nil { - return nil, err - } - if ok { - out[key] = k + out := map[string]Kept[T]{} + for { + select { + case <-ctx.Done(): + return nil, fmt.Errorf("reading the bucket: %w", ctx.Err()) + case e, open := <-w: + if !open { + return nil, errors.New("the bucket stopped answering before it was read whole") + } + if e.Caught { + return out, nil + } + k, err := decode[T](e) + if err != nil { + return nil, err + } + out[e.Key] = k } } - return out, nil } // History is a key's earlier records and its current one, oldest first. @@ -125,8 +154,8 @@ type Chain struct { Assignments Bucket[Assignment] } -// New is a chain over three backends, written only while holds says this controller holds the lease. -func New(builds, rollouts, assignments Backend, holds func() bool) *Chain { +// New is a chain over three backends, written only while holds says this controller may write. A nil holds writes nothing. +func New(builds, rollouts, assignments Backend, holds Holds) *Chain { return &Chain{ Builds: Bucket[Build]{builds, holds}, Rollouts: Bucket[Rollout]{rollouts, holds}, @@ -135,9 +164,9 @@ func New(builds, rollouts, assignments Backend, holds func() bool) *Chain { } // InMemory is a chain held in this process, for tests, written whenever holds says so; nil always holds. -func InMemory(holds func() bool) *Chain { +func InMemory(holds Holds) *Chain { if holds == nil { - holds = func() bool { return true } + holds = func(context.Context) error { return nil } } return New(NewMemory(), NewMemory(), NewMemory(), holds) } diff --git a/internal/chain/chain_test.go b/internal/chain/chain_test.go index bc95b001..4b47260f 100644 --- a/internal/chain/chain_test.go +++ b/internal/chain/chain_test.go @@ -222,14 +222,14 @@ func TestAnAssignmentNamesBuildArtifactCommitAndRollout(t *testing.T) { // the bucket untouched. func TestTheWriterWithoutTheLeaseWritesNothing(t *testing.T) { ctx := context.Background() - held := true + var away error builds, rollouts, assignments := NewMemory(), NewMemory(), NewMemory() - c := New(builds, rollouts, assignments, func() bool { return held }) + c := New(builds, rollouts, assignments, func(context.Context) error { return away }) rev, err := c.Builds.Create(ctx, "b1", Build{ID: "b1"}) if err != nil { t.Fatal(err) } - held = false + away = errors.New("another controller holds the lease") if _, err := c.Builds.Create(ctx, "b2", Build{ID: "b2"}); !errors.Is(err, ErrNotHeld) { t.Fatalf("a create without the lease answered %v", err) } @@ -262,7 +262,7 @@ func TestTheWriterWithoutTheLeaseWritesNothing(t *testing.T) { func TestABucketThatIsAwayIsAnErrorNotAnEmptyOne(t *testing.T) { ctx := context.Background() m := NewMemory() - c := New(m, NewMemory(), NewMemory(), func() bool { return true }) + c := New(m, NewMemory(), NewMemory(), func(context.Context) error { return nil }) if _, err := c.Builds.Create(ctx, "b1", Build{ID: "b1"}); err != nil { t.Fatal(err) } @@ -310,3 +310,27 @@ func TestAWatchHandsOverTheValuesThenEachWrite(t *testing.T) { t.Fatalf("third %+v", e) } } + +// **A key the bus does not take is refused, never rewritten into another** (ADR 0408 §2): a write names the key loudly. +func TestAKeyTheBusTakesNoSpellingOfIsRefused(t *testing.T) { + ctx := context.Background() + c := InMemory(nil) + for _, key := range []string{"", "build one", "build*", ".hidden", "trailing.", "a>b"} { + if _, err := c.Builds.Create(ctx, key, Build{ID: key}); !errors.Is(err, ErrBadKey) { + t.Errorf("the key %q answered %v", key, err) + } + } + for _, key := range []string{"build-1700000000000000000", "anchor.notes", "a/b_c=d"} { + if _, err := c.Builds.Create(ctx, key, Build{ID: key}); err != nil { + t.Errorf("the key %q answered %v", key, err) + } + } +} + +// An empty bucket is read as empty, and the read ends. +func TestAnEmptyBucketIsReadAsEmpty(t *testing.T) { + all, err := InMemory(nil).Assignments.All(context.Background()) + if err != nil || len(all) != 0 { + t.Fatalf("%v %v", all, err) + } +} diff --git a/internal/inventory/builds.go b/internal/inventory/builds.go index a6b6c955..1fa1c78d 100644 --- a/internal/inventory/builds.go +++ b/internal/inventory/builds.go @@ -9,6 +9,8 @@ import ( "time" "github.com/jackc/pgx/v5" + + "github.com/novox/mesh-controller/internal/chain" ) // What has been built. @@ -62,6 +64,10 @@ type Build struct { Asked time.Time // At is when the outcome was recorded — when it finished, not when it was asked. At time.Time + // ReuseKey is the build's reuse key (novox/hq ADR 0350 §1), and Published what it published: an artifact by its + // digest, or a package release by name and version (ADR 0408 §1). Kept on the bus; a build kept in the table has neither. + ReuseKey string + Published *chain.Published } // AskedOrAt is when the build was asked, or when it was recorded when that is not known — the @@ -128,6 +134,11 @@ func (b Build) Worked() bool { return b.Failed == "" } // whoever asked and once on the exchange when nobody was. Recording both would show one build as // two, and which of the two is real is not a question anybody could answer afterwards. func (i *Inventory) RecordBuild(ctx context.Context, b Build) error { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return err + } else if on { + return i.chainRecordBuild(ctx, c, b) + } made, err := json.Marshal(b.Made) if err != nil { return err @@ -207,6 +218,14 @@ func (i *Inventory) RecordMirrored(ctx context.Context, build, why string, refer // now* after something goes wrong, and *what has happened to this* when deciding whether to // trust it. func (i *Inventory) Builds(ctx context.Context, module string, limit int) ([]Build, error) { + if limit <= 0 { + limit = 20 + } + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainBuildList(ctx, c, module, limit) + } if limit <= 0 { limit = 20 } @@ -249,6 +268,11 @@ func (i *Inventory) Builds(ctx context.Context, module string, limit int) ([]Bui // by when the outcome names no module (novox/hq ADR 0219), and what `rebuild` and `replay` read the // repository, path, ref and commit from. False when no outcome with that id was recorded. func (i *Inventory) BuildByID(ctx context.Context, id string) (Build, bool, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return Build{}, false, err + } else if on { + return i.chainBuildByID(ctx, c, id) + } var b Build var made []byte var asked *time.Time @@ -351,6 +375,11 @@ func registeredOnly(held, registered map[string]string) map[string]string { // edges (ADR 0009). A module whose last build recorded no bases is absent, which is also what a // module standing on nothing looks like: an edge the mesh has not derived is not an edge. func (i *Inventory) BuiltAgainst(ctx context.Context) (map[string][]string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainBuiltAgainst(ctx, c) + } rows, err := i.store.Pool().Query(ctx, `select distinct on (module) module, built_against from build @@ -394,6 +423,11 @@ func (i *Inventory) BuiltAgainst(ctx context.Context) (map[string][]string, erro // none — off the trunk, from a builder that predates it, or of a source not known — leaves its module read // as before: every file of a context, and its own directory or root. func (i *Inventory) ReadRepositories(ctx context.Context) (map[string][]ReadRepository, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainReadRepositories(ctx, c) + } // The newest build of each module whatever its outcome: a failed one newer than the newest that // worked leaves that one's build source stale — the merge it was asked for may have changed the closure // (novox/hq ADR 0267), so its module is read whole until a build works again. @@ -580,6 +614,11 @@ func manifestOrNil(raw []byte) any { // One row per module and commit: a module built twice at the same commit is one fact, and the // row asked last is the one whose artifacts are current (novox/hq 04-ISSUES/219). func (i *Inventory) Announceable(ctx context.Context) ([]Build, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainAnnounceable(ctx, c) + } rows, err := i.store.Pool().Query(ctx, `select distinct on (module, commit_hash) id, repository, ref, module, commit_hash, built_on, failed, made, diff --git a/internal/inventory/builds_test.go b/internal/inventory/builds_test.go index fbdfd08a..e0f53b83 100644 --- a/internal/inventory/builds_test.go +++ b/internal/inventory/builds_test.go @@ -4,6 +4,7 @@ package inventory import ( "context" + "encoding/json" "reflect" "strings" "testing" @@ -215,11 +216,15 @@ func TestABuildsBuildSourceComesBackOverWhatItRead(t *testing.T) { if got := read["build-agent"]; len(got) != 1 || got[0].Paths != nil || got[0].Own { t.Fatalf("a build that said no build source reads as %+v", got) } - var stored string - if err := inv.store.Pool().QueryRow(ctx, `select built_contexts::text from build where id = 'said'`).Scan(&stored); err != nil { + kept, found, err := inv.chain.Builds.Get(ctx, "said") + if err != nil || !found { + t.Fatalf("the build is not on the bus: %v %v", found, err) + } + storedRead, err := json.Marshal(kept.Value.Read) + if err != nil { t.Fatal(err) } - if strings.Contains(stored, "paths") || strings.Contains(stored, "own") { + if stored := string(storedRead); strings.Contains(stored, "paths") || strings.Contains(stored, "own") { t.Fatalf("the build source was stored among what the build read: %s", stored) } } diff --git a/internal/inventory/chain.go b/internal/inventory/chain.go new file mode 100644 index 00000000..89fac89a --- /dev/null +++ b/internal/inventory/chain.go @@ -0,0 +1,113 @@ +package inventory + +import ( + "context" + "errors" + "fmt" + "sync" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/novox/mesh-controller/internal/chain" +) + +// **The chain moves to the bus kind by kind** (novox/hq ADR 0408 §8, issue 981). The tables of builds, rollouts and +// assignments are copied once into the chain's buckets by the serving controller at its start (MoveChain), which then +// records each kind it moved in chain_moved (migration 0104). **Until a kind is recorded there nothing reads its bucket; +// once it is, nothing reads its tables**: ChainReads answers which, and every method of a kind asks it first. + +// The kinds of the chain, as chain_moved names them. +const ( + KindBuilds = "builds" +) + +// ErrNoChain is a kind moved to the bus read by an inventory that has no chain attached: a process that did not +// open the bus. Refused loudly, never answered from the tables the move left behind. +var ErrNoChain = errors.New("this kind of record lives on the bus, and this process has not opened the bus") + +// movedKinds is which kinds this process has read as moved: once moved, always moved, so it is read from the store +// until it is true and not after. +type movedKinds struct { + mu sync.Mutex + kind map[string]bool + open ChainOpener +} + +// ChainOpener opens the chain on the bus when a process first needs it: a command that reads and never moved anything has +// no bus connection until a kind it reads turns out to be moved. +type ChainOpener func(ctx context.Context) (*chain.Chain, error) + +// UseChainOpener gives the inventory the means to open the chain when it needs it and none is attached. +func (i *Inventory) UseChainOpener(open ChainOpener) { + i.moved.mu.Lock() + i.moved.open = open + i.moved.mu.Unlock() +} + +// UseChain attaches the chain the moved kinds are read and written through. The serving controller attaches the bus's, +// holding the lease for its writes; a test attaches one in memory. +func (i *Inventory) UseChain(c *chain.Chain) { + i.moved.mu.Lock() + defer i.moved.mu.Unlock() + i.chain = c +} + +// attached is the chain this inventory has, nil when none. +func (i *Inventory) attached() *chain.Chain { + i.moved.mu.Lock() + defer i.moved.mu.Unlock() + return i.chain +} + +// ChainReads says whether a kind of the chain is read from its bucket: chain_moved names it. It answers the chain to read +// through, and ErrNoChain when the kind is moved and none is attached. +func (i *Inventory) ChainReads(ctx context.Context, kind string) (*chain.Chain, bool, error) { + i.moved.mu.Lock() + known := i.moved.kind[kind] + i.moved.mu.Unlock() + if !known { + var moved bool + err := i.store.Pool().QueryRow(ctx, `select counts ? $1 from chain_moved`, kind).Scan(&moved) + var pgErr *pgconn.PgError + if errors.Is(err, pgx.ErrNoRows) || (errors.As(err, &pgErr) && pgErr.Code == "42P01") { + // No row, or a store older than the migration that makes the table: nothing moved. + return nil, false, nil + } + if err != nil { + return nil, false, fmt.Errorf("whether %s moved to the bus cannot be read: %w", kind, err) + } + if !moved { + return nil, false, nil + } + i.moved.mu.Lock() + if i.moved.kind == nil { + i.moved.kind = map[string]bool{} + } + i.moved.kind[kind] = true + i.moved.mu.Unlock() + } + i.moved.mu.Lock() + defer i.moved.mu.Unlock() + if i.chain == nil && i.moved.open != nil { + c, err := i.moved.open(ctx) + if err != nil { + return nil, false, fmt.Errorf("%w: %s: %v", ErrNoChain, kind, err) + } + i.chain = c + } + if i.chain == nil { + return nil, false, fmt.Errorf("%w: %s", ErrNoChain, kind) + } + return i.chain, true, nil +} + +// ChainMoved is, per kind moved, how many records the move wrote; empty before any is. +func (i *Inventory) ChainMoved(ctx context.Context) (map[string]int, error) { + var counts map[string]int + err := i.store.Pool().QueryRow(ctx, `select counts from chain_moved`).Scan(&counts) + if errors.Is(err, pgx.ErrNoRows) { + return map[string]int{}, nil + } + return counts, err +} diff --git a/internal/inventory/chain_builds.go b/internal/inventory/chain_builds.go new file mode 100644 index 00000000..3492d285 --- /dev/null +++ b/internal/inventory/chain_builds.go @@ -0,0 +1,536 @@ +package inventory + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// Builds on the bus (novox/hq ADR 0408, issue 1090): once the kind `builds` is moved (chain.go) every build is a key of +// the builds bucket, under its id, and every read below is made from the bucket's records. The tables are read when the +// kind is not moved, by the methods these stand beside. + +// toChainBuild is a build as the bucket keeps it. +func toChainBuild(b Build) chain.Build { + c := chain.Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, On: b.On, + Path: b.Path, Against: b.Against, Mirrored: b.Mirrored, SourceFingerprint: b.SourceFingerprint, + Failed: b.Failed, Asked: b.Asked.UTC(), At: b.At.UTC(), ReuseKey: b.ReuseKey, Published: b.Published} + if len(b.Manifest) > 0 { + c.Manifest = json.RawMessage(b.Manifest) + } + for _, r := range b.Read { + c.Read = append(c.Read, chain.ReadRepository{Repository: r.Repository, Ref: r.Ref, Paths: r.Paths, Own: r.Own}) + } + for _, s := range b.Sources { + c.Sources = append(c.Sources, chain.Source{Repository: s.Repository, Ref: s.Ref, Paths: s.Paths}) + } + for _, a := range b.Made { + c.Made = append(c.Made, chain.Artifact{Name: a.Name, Kind: a.Kind, Reference: a.Reference}) + } + if c.Published == nil { + c.Published = publishedOf(b.Made) + } + return c +} + +// fromChainBuild is a build as the mesh reads it. +func fromChainBuild(c chain.Build) Build { + b := Build{ID: c.ID, Repository: c.Repository, Ref: c.Ref, Module: c.Module, Commit: c.Commit, On: c.On, Path: c.Path, + Manifest: []byte(c.Manifest), Against: c.Against, Mirrored: c.Mirrored, SourceFingerprint: c.SourceFingerprint, + Failed: c.Failed, Asked: c.Asked, At: c.At, ReuseKey: c.ReuseKey, Published: c.Published} + if len(b.Manifest) == 0 { + b.Manifest = nil + } + for _, r := range c.Read { + b.Read = append(b.Read, ReadRepository{Repository: r.Repository, Ref: r.Ref, Paths: r.Paths, Own: r.Own}) + } + for _, s := range c.Sources { + b.Sources = append(b.Sources, BuildSource{Repository: s.Repository, Ref: s.Ref, Paths: s.Paths}) + } + for _, a := range c.Made { + b.Made = append(b.Made, Artifact{Name: a.Name, Kind: a.Kind, Reference: a.Reference}) + } + return b +} + +// publishedOf is what a build published, read from what it made: a package by name and version, and an artifact by the +// digest of the first thing it made that has one (`@sha256:…`). Nil for a build that published nothing. +func publishedOf(made []Artifact) *chain.Published { + var out chain.Published + for _, a := range made { + switch { + case a.Kind == "package": + if name, version, ok := strings.Cut(a.Reference, "@"); ok && name != "" && version != "" && out.Package == nil { + out.Package = &chain.Package{Name: name, Version: version} + } + case out.Artifact == "": + if _, digest, ok := strings.Cut(a.Reference, "@"); ok && strings.HasPrefix(digest, "sha256:") { + out.Artifact = digest + } + } + } + if out.Artifact == "" && out.Package == nil { + return nil + } + return &out +} + +// chainBuildKey is a build's key. An id the bus takes no spelling of is refused by the write, loudly. +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. +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) + } + return i.RecordMirrored(ctx, b.ID, "", b.Mirrored) +} + +// 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) + if err != nil { + return nil, fmt.Errorf("the builds cannot be read from the bus: %w", err) + } + out := make([]Build, 0, len(all)) + for _, k := range all { + out = append(out, fromChainBuild(k.Value)) + } + return out, nil +} + +// newestRequestOrder sorts builds as newestRequestFirst does: the newest request first, then the newest record. +func newestRequestOrder(builds []Build) { + sort.SliceStable(builds, func(a, b int) bool { + x, y := builds[a].AskedOrAt(), builds[b].AskedOrAt() + if !x.Equal(y) { + return x.After(y) + } + if !builds[a].At.Equal(builds[b].At) { + return builds[a].At.After(builds[b].At) + } + return builds[a].ID > builds[b].ID + }) +} + +// byModule sorts builds by module, then newest request first. +func byModule(builds []Build) { + newestRequestOrder(builds) + sort.SliceStable(builds, func(a, b int) bool { return builds[a].Module < builds[b].Module }) +} + +// worked is the builds that made something for a module: succeeded, and knew what they built. +func worked(all []Build) []Build { + var out []Build + for _, b := range all { + if b.Failed == "" && b.Module != "" { + out = append(out, b) + } + } + return out +} + +func (i *Inventory) chainBuildList(ctx context.Context, c *chain.Chain, module string, limit int) ([]Build, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + var out []Build + for _, b := range all { + if module == "" || b.Module == module { + out = append(out, Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, + On: b.On, Failed: b.Failed, Made: b.Made, Asked: b.Asked, At: b.At}) + } + } + sort.SliceStable(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 + }) + if len(out) > limit { + out = out[:limit] + } + return out, nil +} + +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 { + return Build{}, false, err + } + b := fromChainBuild(k.Value) + return Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, On: b.On, + Failed: b.Failed, Made: b.Made, Path: b.Path, Asked: b.Asked, At: b.At}, true, nil +} + +func (i *Inventory) chainBuiltAgainst(ctx context.Context, c *chain.Chain) (map[string][]string, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + ok := worked(all) + byModule(ok) + out := map[string][]string{} + seen := map[string]bool{} + for _, b := range ok { + if seen[b.Module] { + continue + } + seen[b.Module] = true + if len(b.Against) > 0 { + out[b.Module] = b.Against + } + } + return out, nil +} + +func (i *Inventory) chainReadRepositories(ctx context.Context, c *chain.Chain) (map[string][]ReadRepository, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + // The newest build of each module whatever its outcome: a failed one newer than the newest that worked leaves that + // one's build source stale. + type newestOf struct { + at time.Time + failed bool + } + newest := map[string]newestOf{} + tried := make([]Build, 0, len(all)) + for _, b := range all { + if b.Module != "" { + tried = append(tried, b) + } + } + byModule(tried) + for _, b := range tried { + if _, seen := newest[b.Module]; !seen { + newest[b.Module] = newestOf{b.AskedOrAt(), b.Failed != ""} + } + } + owns, err := i.moduleSources(ctx) + if err != nil { + return nil, err + } + ok := worked(all) + byModule(ok) + read := map[string][]ReadRepository{} + seen := map[string]bool{} + for _, b := range ok { + if seen[b.Module] { + continue + } + seen[b.Module] = true + built := b.AskedOrAt() + of := append([]ReadRepository(nil), b.Read...) + sources := b.Sources + whole := "" + if n, known := newest[b.Module]; known && n.failed && n.at.After(built) && len(sources) > 0 { + sources, whole = nil, "a newer build of it failed" + } + of = WithBuildSources(of, sources) + if own, has := owns[b.Module]; has { + of = OwnRepositoryAsOwn(of, own) + } + if whole != "" { + of = append(of, ReadRepository{Own: true, Whole: whole}) + } + if len(of) > 0 { + for k := range of { + of[k].Built, of[k].Looked = built, newest[b.Module].at + } + read[b.Module] = of + } + } + return read, nil +} + +// moduleSources is the repository each registered module is registered from. +func (i *Inventory) moduleSources(ctx context.Context) (map[string]string, error) { + rows, err := i.store.Pool().Query(ctx, `select name, source from module where source is not null`) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]string{} + for rows.Next() { + var name, source string + if err := rows.Scan(&name, &source); err != nil { + return nil, err + } + out[name] = source + } + return out, rows.Err() +} + +func (i *Inventory) chainAnnounceable(ctx context.Context, c *chain.Chain) ([]Build, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + var ok []Build + for _, b := range worked(all) { + if b.Commit != "" { + ok = append(ok, b) + } + } + newestRequestOrder(ok) + type key struct{ module, commit string } + seen := map[key]bool{} + var out []Build + for _, b := range ok { + k := key{b.Module, b.Commit} + if seen[k] { + continue + } + seen[k] = true + out = append(out, Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, + On: b.On, Failed: b.Failed, Made: b.Made, Path: b.Path, Manifest: b.Manifest, Against: b.Against, At: b.At}) + } + sort.SliceStable(out, func(a, b int) bool { return out[a].At.Before(out[b].At) }) + return out, nil +} + +// failedAtAGate is the builds that failed their gate. The gate's verdicts stay in their table (issue 1097). +func (i *Inventory) failedAtAGate(ctx context.Context) (map[string]bool, error) { + rows, err := i.store.Pool().Query(ctx, `select build from build_gate where verdict = 'failed'`) + if err != nil { + return nil, err + } + defer rows.Close() + out := map[string]bool{} + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, err + } + out[id] = true + } + return out, rows.Err() +} + +func (i *Inventory) chainSourceRows(ctx context.Context, c *chain.Chain, module string) (map[string][]sourceRow, []string, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, nil, err + } + failed, err := i.failedAtAGate(ctx) + if err != nil { + return nil, nil, err + } + ok := worked(all) + byModule(ok) + out := map[string][]sourceRow{} + var order []string + for _, b := range ok { + if module != "" && b.Module != module { + continue + } + made, err := json.Marshal(b.Made) + if err != nil { + return nil, nil, err + } + if _, seen := out[b.Module]; !seen { + order = append(order, b.Module) + } + out[b.Module] = append(out[b.Module], sourceRow{id: b.ID, commit: b.Commit, fingerprint: b.SourceFingerprint, + made: made, manifest: b.Manifest, failedGate: failed[b.ID]}) + } + return out, order, nil +} + +// canonical is a JSON document encoded the one way (objects' keys sorted, compact), so the same content hashes alike +// however the bucket's record spells it. +func canonicalJSON(raw []byte) []byte { + if len(bytes.TrimSpace(raw)) == 0 { + return nil + } + out, err := canonical(raw) + if err != nil { + return raw + } + return out +} + +// chainFingerprint is what a build would put on a machine, hashed: its artifacts and its manifest. +func chainFingerprint(b Build) string { + made, _ := json.Marshal(b.Made) + sum := sha256.Sum256(append(append(canonicalJSON(made), 0), canonicalJSON(b.Manifest)...)) + return hex.EncodeToString(sum[:]) +} + +func (i *Inventory) chainBuildFingerprints(ctx context.Context, c *chain.Chain, module string) (map[string]string, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + var ok []Build + for _, b := range all { + if b.Module == module && b.Failed == "" && b.Commit != "" { + ok = append(ok, b) + } + } + sortByRecorded(ok) + if len(ok) > 50 { + ok = ok[:50] + } + out := map[string]string{} + for _, b := range ok { + if _, seen := out[b.Commit]; !seen { + out[b.Commit] = chainFingerprint(b) + } + } + return out, nil +} + +func (i *Inventory) chainFingerprints(ctx context.Context, c *chain.Chain) (map[string]map[string]string, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + var ok []Build + for _, b := range all { + if b.Module != "" && b.Failed == "" && b.Commit != "" { + ok = append(ok, b) + } + } + sortByRecorded(ok) + out := map[string]map[string]string{} + for _, b := range ok { + if out[b.Module] == nil { + out[b.Module] = map[string]string{} + } + if _, seen := out[b.Module][b.Commit]; !seen { + out[b.Module][b.Commit] = chainFingerprint(b) + } + } + return out, nil +} + +// sortByRecorded is newest record first, as `order by at desc`. +func sortByRecorded(builds []Build) { + sort.SliceStable(builds, func(a, b int) bool { + if !builds[a].At.Equal(builds[b].At) { + return builds[a].At.After(builds[b].At) + } + return builds[a].ID > builds[b].ID + }) +} + +func (i *Inventory) chainBuildOf(ctx context.Context, c *chain.Chain, module, commit string) (string, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return "", err + } + var ok []Build + for _, b := range all { + if b.Module == module && b.Commit == commit && b.Failed == "" { + ok = append(ok, b) + } + } + sortByRecorded(ok) + if len(ok) == 0 { + return "", nil + } + return ok[0].ID, nil +} + +// chainManifestBuild is the newest successful build of a module from a commit (or one it abbreviates) that kept a manifest. +func (i *Inventory) chainManifestBuild(ctx context.Context, c *chain.Chain, module, commit string) (string, []byte, bool, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return "", nil, false, err + } + var ok []Build + for _, b := range all { + if b.Module == module && b.Failed == "" && len(b.Manifest) > 0 && string(bytes.TrimSpace(b.Manifest)) != "null" && + (b.Commit == commit || strings.HasPrefix(b.Commit, commit)) { + ok = append(ok, b) + } + } + sortByRecorded(ok) + if len(ok) == 0 { + return "", nil, false, nil + } + return ok[0].ID, ok[0].Manifest, true, nil +} + +// chainManifestOf is the manifest a build was recorded with. +func (i *Inventory) chainManifestOf(ctx context.Context, c *chain.Chain, id string) ([]byte, error) { + k, ok, err := c.Builds.Get(ctx, chainBuildKey(id)) + if err != nil || !ok { + return nil, err + } + return []byte(k.Value.Manifest), nil +} + +// chainRecentMade is the artifacts and bases of the KeptBuilds newest successful builds of each registered module. +func (i *Inventory) chainRecentMade(ctx context.Context, c *chain.Chain) ([]Build, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + rows, err := i.store.Pool().Query(ctx, `select name from module`) + if err != nil { + return nil, err + } + registered := map[string]bool{} + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + rows.Close() + return nil, err + } + registered[name] = true + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + var ok []Build + for _, b := range worked(all) { + if registered[b.Module] { + ok = append(ok, b) + } + } + sortByRecorded(ok) + back := map[string]int{} + var out []Build + for _, b := range ok { + back[b.Module]++ + if back[b.Module] <= KeptBuilds { + out = append(out, b) + } + } + return out, nil +} + +// chainEveryReferenceMade is every successful build's artifacts, oldest build first. +func (i *Inventory) chainEveryReferenceMade(ctx context.Context, c *chain.Chain) ([]Build, error) { + all, err := chainBuilds(ctx, c) + if err != nil { + return nil, err + } + ok := worked(all) + sort.SliceStable(ok, func(a, b int) bool { + if !ok[a].At.Equal(ok[b].At) { + return ok[a].At.Before(ok[b].At) + } + return ok[a].ID < ok[b].ID + }) + return ok, nil +} diff --git a/internal/inventory/chain_builds_test.go b/internal/inventory/chain_builds_test.go new file mode 100644 index 00000000..8cee9d17 --- /dev/null +++ b/internal/inventory/chain_builds_test.go @@ -0,0 +1,100 @@ +package inventory + +import ( + "encoding/json" + "reflect" + "testing" + "time" + + "github.com/novox/mesh-controller/internal/chain" +) + +// **A build keeps its reuse key and what it published, and comes back whole from the bucket** (novox/hq ADR 0408 §1, +// issue 1090): an artifact by the digest of what it made, a package release by name and version. +func TestABuildKeepsItsReuseKeyAndWhatItPublishedWhenKeptOnTheBus(t *testing.T) { + at := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + b := Build{ID: "build-1", Repository: "novox/notes", Ref: "main", Module: "notes", Commit: "c1", On: "ace", Path: "modules/notes", + Manifest: []byte(`{"module":"notes"}`), Against: []string{"artifact-store://base@sha256:aa"}, + Read: []ReadRepository{{Repository: "novox/other", Ref: "main"}}, + Sources: []BuildSource{{Paths: []string{"modules/notes/module.json"}}}, + SourceFingerprint: "fp", Asked: at, At: at.Add(time.Minute), ReuseKey: "rk-1", + Made: []Artifact{{Name: "notes", Kind: "image", Reference: "artifact-store://notes@sha256:bb"}, + {Name: "sdk", Kind: "package", Reference: "mesh-sdk@1.4.0"}}} + kept := toChainBuild(b) + if kept.ReuseKey != "rk-1" { + t.Errorf("the reuse key is %q", kept.ReuseKey) + } + if kept.Published == nil || kept.Published.Artifact != "sha256:bb" || kept.Published.Package == nil || + kept.Published.Package.Name != "mesh-sdk" || kept.Published.Package.Version != "1.4.0" { + t.Errorf("it published %+v", kept.Published) + } + // Through the bucket's bytes and back: nothing is lost. + raw, err := json.Marshal(kept) + if err != nil { + t.Fatal(err) + } + var back chain.Build + if err := json.Unmarshal(raw, &back); err != nil { + t.Fatal(err) + } + got := fromChainBuild(back) + b.Published = kept.Published + if !reflect.DeepEqual(got, b) { + t.Fatalf("the build came back as\n %+v\nnot\n %+v", got, b) + } +} + +func TestWhatABuildPublishedIsReadFromWhatItMade(t *testing.T) { + for name, tc := range map[string]struct { + made []Artifact + want *chain.Published + }{ + "nothing": {nil, nil}, + "an archive with no digest": {[]Artifact{{Name: "config", Kind: "archive", Reference: "…/blobs/sha256:…"}}, nil}, + "an image": {[]Artifact{{Name: "a", Kind: "image", Reference: "artifact-store://a@sha256:11"}}, + &chain.Published{Artifact: "sha256:11"}}, + "the first digest stands": {[]Artifact{{Kind: "image", Reference: "x@sha256:11"}, {Kind: "archive", Reference: "y@sha256:22"}}, + &chain.Published{Artifact: "sha256:11"}}, + "a package": {[]Artifact{{Name: "sdk", Kind: "package", Reference: "mesh-sdk@2.0.1"}}, + &chain.Published{Package: &chain.Package{Name: "mesh-sdk", Version: "2.0.1"}}}, + "a package with no version": {[]Artifact{{Name: "sdk", Kind: "package", Reference: "mesh-sdk"}}, nil}, + } { + if got := publishedOf(tc.made); !reflect.DeepEqual(got, tc.want) { + t.Errorf("%s: published %+v, not %+v", name, got, tc.want) + } + } +} + +// **The same content hashes alike, however the record spells it** (issue 280's fingerprints, now over the bucket's bytes +// and not the store's jsonb text): key order and white space of a manifest change nothing. +func TestTheSameBuildHashesAlikeWhateverTheSpellingOfItsRecord(t *testing.T) { + a := Build{Manifest: []byte(`{"module": "notes", "version": "1"}`), Made: []Artifact{{Name: "n", Kind: "image", Reference: "r"}}} + b := Build{Manifest: []byte("{\"version\":\"1\",\n \"module\":\"notes\"}"), Made: []Artifact{{Name: "n", Kind: "image", Reference: "r"}}} + if chainFingerprint(a) != chainFingerprint(b) { + t.Fatal("two spellings of one manifest hash apart") + } + b.Made[0].Reference = "other" + if chainFingerprint(a) == chainFingerprint(b) { + t.Fatal("two artifacts hash alike") + } +} + +// **Newest request first, as the store's ordering was**: the newest ask, then the newest record, then the newest id. +func TestBuildsAreOrderedNewestRequestFirst(t *testing.T) { + t0 := time.Date(2026, 10, 18, 3, 0, 0, 0, time.UTC) + builds := []Build{ + {ID: "old", Asked: t0, At: t0.Add(time.Hour)}, + {ID: "unasked", At: t0.Add(2 * time.Hour)}, + {ID: "new", Asked: t0.Add(3 * time.Hour), At: t0.Add(3*time.Hour + time.Minute)}, + {ID: "same-a", Asked: t0.Add(4 * time.Hour), At: t0.Add(5 * time.Hour)}, + {ID: "same-b", Asked: t0.Add(4 * time.Hour), At: t0.Add(5 * time.Hour)}, + } + newestRequestOrder(builds) + var got []string + for _, b := range builds { + got = append(got, b.ID) + } + if want := []string{"same-b", "same-a", "new", "unasked", "old"}; !reflect.DeepEqual(got, want) { + t.Fatalf("ordered %v, not %v", got, want) + } +} diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go new file mode 100644 index 00000000..123fa01b --- /dev/null +++ b/internal/inventory/chain_move.go @@ -0,0 +1,265 @@ +package inventory + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + + "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/chain" +) + +// **The move** (novox/hq ADR 0408 §8, issue 981 note 7 Q2): the serving controller writes the rows of a kind into its +// bucket at its start, holding the lease and after the buckets exist, and records the kind in chain_moved. Idempotent: a +// key that exists is already moved and is never overwritten, so a bucket value written since is not laid over by the row +// it came from. It counts the rows against the keys, logs the largest value, and **refuses** (a Refusal, which the +// controller raises as an urgent condition) on a count that differs or a value over the bus's largest, with nothing +// switched: a kind is recorded moved only once all of it is in the bucket. + +// Refusal is a move that stopped without switching a kind, and why. It is permanent until the cause is put right; any +// other error of a move (the bus away, a write not granted) is tried again. +type Refusal struct { + Kind, Why string +} + +func (r *Refusal) Error() string { + return fmt.Sprintf("the move of %s to the bus is refused: %s", r.Kind, r.Why) +} + +// MoveReport is what a move of one kind did. +type MoveReport struct { + Kind string + // 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 + // Largest is the biggest value, in bytes. + Largest int + // Already is a kind an earlier start moved: nothing was done. + Already bool +} + +// 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) + // 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 func(ctx context.Context, tx pgx.Tx) (int, error) + // lock keeps the tables from taking a row while the switch is recorded. + lock string +} + +type moveRow struct { + key string + value any + size int +} + +func (i *Inventory) movers() []mover { + return []mover{{ + kind: KindBuilds, + rows: func(ctx context.Context) ([]moveRow, error) { + builds, err := i.allBuildRows(ctx) + if err != nil { + return nil, err + } + out := make([]moveRow, 0, len(builds)) + 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)}) + } + return out, nil + }, + create: func(ctx context.Context, c *chain.Chain, row moveRow) error { + _, err := c.Builds.Create(ctx, row.key, row.value.(chain.Build)) + return err + }, + 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) + return n, err + }, + lock: `lock table build in share mode`, + }} +} + +// MoveChain moves every kind not moved yet, in order, and says what it did. It needs the chain attached and the lease +// held (the chain's writes ask). It stops at the first kind that fails. +func (i *Inventory) MoveChain(ctx context.Context, say func(format string, args ...any)) ([]MoveReport, error) { + if say == nil { + say = func(string, ...any) {} + } + if i.attached() == nil { + return nil, ErrNoChain + } + var out []MoveReport + for _, m := range i.movers() { + r, err := i.moveKind(ctx, m, say) + if err != nil { + return out, err + } + out = append(out, r) + } + return out, nil +} + +// movePasses is how often a move reads the tables again when they took a row while it wrote: the controller before this +// one may still have been writing as the lease changed hands. +const movePasses = 5 + +func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...any)) (MoveReport, error) { + report := MoveReport{Kind: m.kind} + moved, err := i.ChainMoved(ctx) + if err != nil { + return report, err + } + if _, done := moved[m.kind]; done { + report.Already = true + return report, nil + } + for pass := 1; pass <= movePasses; pass++ { + rows, 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 + // 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 { + report.Largest = max(report.Largest, r.size) + if r.size > broker.ChainValueBytes { + return report, &Refusal{m.kind, fmt.Sprintf("the value of %s is %d bytes and the bus holds at most %d; "+ + "nothing is cut, so the record must be made smaller first", r.key, r.size, broker.ChainValueBytes)} + } + } + report.Written, report.Present = 0, 0 + for _, r := range rows { + switch err := m.create(ctx, i.attached(), r); { + case err == nil: + report.Written++ + case errors.Is(err, chain.ErrExists): + report.Present++ + default: + return report, fmt.Errorf("%s %s is not written to the bus: %w", m.kind, r.key, err) + } + } + // Rows against keys: every row's key is in the bucket. + keys, err := m.keys(ctx, i.attached()) + if err != nil { + return report, fmt.Errorf("the keys of the %s cannot be read back: %w", m.kind, err) + } + have := make(map[string]bool, len(keys)) + for _, k := range keys { + have[k] = true + } + for _, r := range rows { + if !have[r.key] { + return report, &Refusal{m.kind, fmt.Sprintf("the tables hold %d records and the bucket does not hold %s", len(rows), r.key)} + } + } + settled, err := i.switchKind(ctx, m, report) + if err != nil { + return report, err + } + if settled { + say("moved the %s to the bus: %d records, %d written now and %d already there, the largest %d bytes", + m.kind, report.Rows, report.Written, report.Present, report.Largest) + return report, nil + } + say("the %s took a row while they were moved, so they are read again (pass %d of %d)", m.kind, pass, movePasses) + } + return report, &Refusal{m.kind, fmt.Sprintf("the tables kept taking rows through %d passes; another controller is still writing them", movePasses)} +} + +// switchKind records the kind as moved, in one transaction that holds the tables against writes and finds their count +// as the move read it: false when they took a row since. +func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool, error) { + tx, err := i.store.Pool().Begin(ctx) + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback(ctx) }() + if _, err := tx.Exec(ctx, m.lock); err != nil { + return false, err + } + n, err := m.count(ctx, tx) + if err != nil { + return false, err + } + if n != r.Rows { + return false, nil + } + 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, + `insert into chain_moved (counts, largest) values ($1, $2) + on conflict (only_one) do update + set counts = chain_moved.counts || excluded.counts, largest = chain_moved.largest || excluded.largest, + moved_at = now()`, counts, largest); err != nil { + return false, err + } + if err := tx.Commit(ctx); err != nil { + return false, err + } + i.moved.mu.Lock() + if i.moved.kind == nil { + i.moved.kind = map[string]bool{} + } + i.moved.kind[m.kind] = true + i.moved.mu.Unlock() + return true, nil +} + +// allBuildRows is every build of the table, whole. +func (i *Inventory) allBuildRows(ctx context.Context) ([]Build, error) { + rows, err := i.store.Pool().Query(ctx, + `select id, repository, ref, coalesce(module, ''), commit_hash, built_on, failed, made, source_path, manifest, + built_against, built_contexts, build_sources, source_fingerprint, asked, at + from build order by at, id`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Build + for rows.Next() { + var b Build + var made, manifest, against, read, sources []byte + var asked *time.Time + if err := rows.Scan(&b.ID, &b.Repository, &b.Ref, &b.Module, &b.Commit, &b.On, &b.Failed, &made, &b.Path, + &manifest, &against, &read, &sources, &b.SourceFingerprint, &asked, &b.At); err != nil { + return nil, err + } + if asked != nil { + b.Asked = *asked + } + b.Manifest = manifest + for _, part := range []struct { + raw []byte + to any + who string + }{{made, &b.Made, "made"}, {against, &b.Against, "built_against"}, {read, &b.Read, "built_contexts"}, + {sources, &b.Sources, "build_sources"}} { + if len(part.raw) == 0 { + continue + } + if err := json.Unmarshal(part.raw, part.to); err != nil { + return nil, fmt.Errorf("build %s: %s is not readable: %w", b.ID, part.who, err) + } + } + out = append(out, b) + } + return out, rows.Err() +} diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go new file mode 100644 index 00000000..2a4cc219 --- /dev/null +++ b/internal/inventory/chain_move_test.go @@ -0,0 +1,272 @@ +//go:build integration + +package inventory + +import ( + "context" + "errors" + "reflect" + "testing" + + "github.com/novox/mesh-controller/internal/chain" +) + +// A build row as the store held it before the move: written by SQL, the only hand-written state these tests have, for +// it stands for what the mesh held before this release. +func aBuildRow(t *testing.T, inv *Inventory, id, module, commit, failed string) { + t.Helper() + inv.ExecForTest(t, + `insert into build (id, repository, ref, module, commit_hash, built_on, failed, made, source_path, manifest, + built_against, built_contexts, asked, source_fingerprint, build_sources) + values ($1::text, 'novox/' || $2::text, 'main', $2::text, $3::text, 'ace', $4::text, + case when $4::text = '' + then ('[{"name":"app","kind":"image","reference":"artifact-store://' || $2::text || '@sha256:' || $1::text || '"}]')::jsonb + else '[]'::jsonb end, + 'modules/' || $2::text, ('{"module":"' || $2::text || '"}')::jsonb, + '["artifact-store://base@sha256:aa"]'::jsonb, '[{"repository":"novox/other","ref":"main"}]'::jsonb, + now() - interval '1 hour', 'fp-' || $1::text, + ('[{"paths":["modules/' || $2::text || '/module.json"]}]')::jsonb)`, + id, module, commit, failed) +} + +func rowsInTheTable(t *testing.T, inv *Inventory, table string) string { + t.Helper() + rows := inv.QueryForTest(t, `select count(*) from `+table) + return rows[0][0] +} + +// **The move copies every build row into the builds bucket, and a second run writes nothing** (ADR 0408 §8, issue +// 1090): each row under its id, whole; the kind recorded in chain_moved; and from then on a build is read from the +// bucket and written to it, the table taking no more. +func TestTheMoveCopiesTheBuildRowsAndAgainWritesNothing(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + aBuildRow(t, inv, "build-2", "notes", "c2", "") + aBuildRow(t, inv, "build-3", "wiki", "", "compile error") + + // Before the move the tables are read, and the bucket is empty. + if before, err := inv.Builds(ctx, "", 10); err != nil || len(before) != 3 { + t.Fatalf("before the move the tables hold %d builds: %v", len(before), err) + } + if moved, _ := inv.ChainMoved(ctx); len(moved) != 0 { + t.Fatalf("moved before the move: %v", moved) + } + + reports, err := inv.MoveChain(ctx, nil) + 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) + } + moved, err := inv.ChainMoved(ctx) + if err != nil || moved[KindBuilds] != 3 { + t.Fatalf("chain_moved says %v: %v", moved, err) + } + kept, found, err := inv.chain.Builds.Get(ctx, "build-2") + if err != nil || !found { + t.Fatalf("build-2 is not in the bucket: %v %v", found, err) + } + if v := kept.Value; v.Module != "notes" || v.Commit != "c2" || v.Repository != "novox/notes" || v.Path != "modules/notes" || + v.SourceFingerprint != "fp-build-2" || len(v.Made) != 1 || len(v.Against) != 1 || len(v.Read) != 1 || len(v.Sources) != 1 || + v.Published == nil || v.Published.Artifact != "sha256:build-2" || v.Asked.IsZero() || v.At.IsZero() || len(v.Manifest) == 0 { + t.Fatalf("build-2 was moved as %+v", v) + } + + // From now on the bucket is what is read and written. + if err := inv.RecordBuild(ctx, aBuild("build-4", "notes", "")); err != nil { + t.Fatal(err) + } + if n := rowsInTheTable(t, inv, "build"); n != "3" { + t.Fatalf("the table took a build after the move: %s rows", n) + } + after, err := inv.Builds(ctx, "notes", 10) + if err != nil || len(after) != 3 { + t.Fatalf("notes has %d builds on the bus: %v", len(after), err) + } + + // 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) + } + 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) + } + } +} + +// **A move that stopped half way is finished by the next start, and never lays a row over a newer bucket value** (ADR +// 0408 §8): the keys that exist are present, the others are written. +func TestAControllerRestartedMidMoveFinishesItAndOverwritesNothing(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + aBuildRow(t, inv, "build-2", "notes", "c2", "") + aBuildRow(t, inv, "build-3", "wiki", "c3", "") + // An earlier start wrote build-1, and the bucket has had a newer value since. + newer := chain.Build{ID: "build-1", Repository: "novox/notes", Module: "notes", Commit: "newer"} + if _, err := inv.chain.Builds.Create(ctx, "build-1", newer); err != nil { + t.Fatal(err) + } + reports, err := inv.MoveChain(ctx, nil) + 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]) + } + got, _, _ := inv.chain.Builds.Get(ctx, "build-1") + if !reflect.DeepEqual(got.Value, newer) { + t.Fatalf("the newer value was laid over: %+v", got.Value) + } + if moved, _ := inv.ChainMoved(ctx); moved[KindBuilds] != 3 { + t.Fatalf("chain_moved says %v", moved) + } +} + +// **A value over the bus's largest message stops the move with nothing written and nothing switched** (ADR 0408 §6): +// a Refusal, no key in the bucket, no kind recorded, and the tables still what is read. +func TestTheMoveRefusesAValueOverTheLargestMessage(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + inv.ExecForTest(t, `insert into build (id, repository, module, commit_hash, manifest) + values ('build-big', 'novox/wiki', 'wiki', 'c9', ('{"x":"' || repeat('a', 1100000) || '"}')::jsonb)`) + _, err := inv.MoveChain(ctx, nil) + var refusal *Refusal + if !errors.As(err, &refusal) || refusal.Kind != KindBuilds { + t.Fatalf("the move answered %v", err) + } + if keys, _ := inv.chain.Builds.Keys(ctx); len(keys) != 0 { + t.Fatalf("the bucket holds %v after a refused move", keys) + } + if moved, _ := inv.ChainMoved(ctx); len(moved) != 0 { + t.Fatalf("a refused move recorded %v", moved) + } + if builds, err := inv.Builds(ctx, "", 10); err != nil || len(builds) != 2 { + t.Fatalf("the tables are no longer read: %d builds, %v", len(builds), err) + } +} + +// losing is a bucket that says it wrote a key and kept nothing: the move's count of keys against rows must see it. +type losing struct { + chain.Backend + lose string +} + +func (l losing) Create(ctx context.Context, key string, value []byte) (uint64, error) { + if key == l.lose { + return 1, nil + } + return l.Backend.Create(ctx, key, value) +} + +// **The move refuses on a count that differs** (ADR 0408 §8): a row whose key is not in the bucket afterwards stops it, +// nothing is recorded, and the tables are still read. +func TestTheMoveRefusesOnACountThatDiffers(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + aBuildRow(t, inv, "build-2", "notes", "c2", "") + inv.UseChain(chain.New(losing{chain.NewMemory(), "build-2"}, chain.NewMemory(), chain.NewMemory(), + func(context.Context) error { return nil })) + _, err := inv.MoveChain(ctx, nil) + var refusal *Refusal + if !errors.As(err, &refusal) { + t.Fatalf("the move answered %v", err) + } + if moved, _ := inv.ChainMoved(ctx); len(moved) != 0 { + t.Fatalf("a refused move recorded %v", moved) + } +} + +// **Without the lease the move writes nothing**, and fails the way the bus being away does — tried again, not refused: +// the tables are still read. +func TestTheMoveWithoutTheLeaseWritesNothing(t *testing.T) { + inv, finish := ForTestBefore(t, 105) + finish() + ctx := context.Background() + aBuildRow(t, inv, "build-1", "notes", "c1", "") + mem := chain.NewMemory() + inv.UseChain(chain.New(mem, chain.NewMemory(), chain.NewMemory(), + func(context.Context) error { return errors.New("another controller holds the lease") })) + _, err := inv.MoveChain(ctx, nil) + var refusal *Refusal + if err == nil || errors.As(err, &refusal) || !errors.Is(err, chain.ErrNotHeld) { + t.Fatalf("the move answered %v", err) + } + if mem.Writes() != 0 { + t.Fatalf("the bucket took %d writes without the lease", mem.Writes()) + } + if moved, _ := inv.ChainMoved(ctx); len(moved) != 0 { + t.Fatalf("recorded %v", moved) + } +} + +// **Once builds are on the bus every reader of them reads the bucket** (issue 1090): a build recorded is the key of its +// id, recorded again it stays one, and the lists the planner reads are made from the bucket — the table takes none. +func TestBuildsOnTheBusAreReadBackByEveryReader(t *testing.T) { + inv := fresh(t) + ctx := context.Background() + first := aBuild("build-1", "notes", "") + first.Commit = "c1" + first.Against = []string{"artifact-store://base@sha256:aa"} + first.ReuseKey = "rk-1" + second := aBuild("build-2", "notes", "") + second.Commit = "c2" + failed := aBuild("build-3", "notes", "compile error") + other := aBuild("build-4", "wiki", "") + for _, b := range []Build{first, second, failed, other, first} { + if err := inv.RecordBuild(ctx, b); err != nil { + t.Fatal(err) + } + } + if n := rowsInTheTable(t, inv, "build"); n != "0" { + t.Fatalf("the table took %s builds", n) + } + kept, found, err := inv.chain.Builds.Get(ctx, "build-1") + if err != nil || !found || kept.Value.ReuseKey != "rk-1" { + t.Fatalf("build-1 on the bus: %+v %v %v", kept.Value, found, err) + } + if history, _ := inv.chain.Builds.History(ctx, "build-1"); len(history) != 1 { + t.Fatalf("a build recorded twice has %d values", len(history)) + } + list, err := inv.Builds(ctx, "notes", 10) + if err != nil || len(list) != 3 { + t.Fatalf("notes has %d builds: %v", len(list), err) + } + one, found, err := inv.BuildByID(ctx, "build-2") + if err != nil || !found || one.Commit != "c2" || one.Module != "notes" { + t.Fatalf("build-2 reads as %+v %v %v", one, found, err) + } + if _, found, _ := inv.BuildByID(ctx, "build-nope"); found { + t.Fatal("a build nobody recorded was found") + } + if id, err := inv.BuildOf(ctx, "notes", "c1"); err != nil || id != "build-1" { + t.Fatalf("the build of notes at c1 is %q: %v", id, err) + } + held, err := inv.Held(ctx) + if err != nil || held["notes/config"] == "" || held["wiki/config"] == "" { + t.Fatalf("held %v: %v", held, err) + } + against, err := inv.BuiltAgainst(ctx) + if err != nil || len(against["notes"]) != 0 { + // The newest successful build of notes, build-2, stood on nothing. + t.Fatalf("built against %v: %v", against, err) + } + announced, err := inv.Announceable(ctx) + if err != nil || len(announced) != 3 { + t.Fatalf("announceable %d: %v", len(announced), err) + } + prints, err := inv.Fingerprints(ctx) + if err != nil || prints["notes"]["c1"] == "" || prints["notes"]["c2"] == "" || prints["wiki"]["c0ffeebuild-4"] == "" { + t.Fatalf("fingerprints %v: %v", prints, err) + } +} diff --git a/internal/inventory/collection.go b/internal/inventory/collection.go index 6e60a872..6c0ba092 100644 --- a/internal/inventory/collection.go +++ b/internal/inventory/collection.go @@ -159,44 +159,23 @@ func (i *Inventory) KeepSet(ctx context.Context) (map[string][]string, error) { // no releases left to return between, and its build rows stay only as history. Without the // join every module ever built kept five builds' artifacts for ever — and once the store's // collector runs for real, what the keep set says is what the disk holds. - recent, err := i.store.Pool().Query(ctx, - `select made, built_against from ( - select b.made, b.built_against, - row_number() over (partition by b.module order by b.at desc, b.id desc) as back - from build b - join module m on m.name = b.module - where b.failed = '' and b.module is not null and b.module <> '' - ) ranked where back <= $1`, KeptBuilds) + recent, err := i.recentBuilds(ctx) if err != nil { return nil, err } - defer recent.Close() var stoodOn []string - for recent.Next() { - var raw, against []byte - if err := recent.Scan(&raw, &against); err != nil { - return nil, err - } - var made []Artifact - if err := json.Unmarshal(raw, &made); err == nil { - for _, a := range made { - if reference := asRecorded(a.Reference); reference != "" { - because(reference, KeptByRecentBuild) - } + for _, r := range recent { + for _, a := range r.made { + if reference := asRecorded(a.Reference); reference != "" { + because(reference, KeptByRecentBuild) } } - var bases []string - if err := json.Unmarshal(against, &bases); err == nil { - for _, b := range bases { - if reference := asRecorded(b); reference != "" { - stoodOn = append(stoodOn, reference) - } + for _, b := range r.against { + if reference := asRecorded(b); reference != "" { + stoodOn = append(stoodOn, reference) } } } - if err := recent.Err(); err != nil { - return nil, err - } // **A copied base is kept by the kept builds that stood on it, and by nothing else** (novox/hq // ADR 0257). It has no builds of its own to count five of, and no machine is ever handed a base, @@ -340,27 +319,97 @@ func (i *Inventory) mirroredSet(ctx context.Context) (map[string]bool, error) { return out, rows.Err() } -// everyReferenceMade is every artifact reference any successful build recorded, oldest build first. -func (i *Inventory) everyReferenceMade(ctx context.Context) ([]string, error) { +// recentBuild is what one of the KeptBuilds newest successful builds of a module made and stood on. +type recentBuild struct { + made []Artifact + against []string +} + +// recentBuilds is, of each module the mesh still holds, its KeptBuilds newest successful builds' artifacts and bases. +func (i *Inventory) recentBuilds(ctx context.Context) ([]recentBuild, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + builds, err := i.chainRecentMade(ctx, c) + if err != nil { + return nil, err + } + out := make([]recentBuild, 0, len(builds)) + for _, b := range builds { + out = append(out, recentBuild{made: b.Made, against: b.Against}) + } + return out, nil + } rows, err := i.store.Pool().Query(ctx, - `select made from build where failed = '' and module is not null and module <> '' - order by at asc, id asc`) + `select made, built_against from ( + select b.made, b.built_against, + row_number() over (partition by b.module order by b.at desc, b.id desc) as back + from build b + join module m on m.name = b.module + where b.failed = '' and b.module is not null and b.module <> '' + ) ranked where back <= $1`, KeptBuilds) if err != nil { return nil, err } defer rows.Close() - seen := map[string]bool{} - var out []string + var out []recentBuild for rows.Next() { - var raw []byte - if err := rows.Scan(&raw); err != nil { + var raw, against []byte + if err := rows.Scan(&raw, &against); err != nil { return nil, err } - var made []Artifact - if err := json.Unmarshal(raw, &made); err != nil { - continue + var r recentBuild + if err := json.Unmarshal(raw, &r.made); err != nil { + r.made = nil } - for _, a := range made { + if err := json.Unmarshal(against, &r.against); err != nil { + r.against = nil + } + out = append(out, r) + } + return out, rows.Err() +} + +// everyReferenceMade is every artifact reference any successful build recorded, oldest build first. +func (i *Inventory) everyReferenceMade(ctx context.Context) ([]string, error) { + var made [][]Artifact + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + builds, err := i.chainEveryReferenceMade(ctx, c) + if err != nil { + return nil, err + } + for _, b := range builds { + made = append(made, b.Made) + } + } else { + rows, err := i.store.Pool().Query(ctx, + `select made from build where failed = '' and module is not null and module <> '' + order by at asc, id asc`) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var raw []byte + if err := rows.Scan(&raw); err != nil { + return nil, err + } + var a []Artifact + if err := json.Unmarshal(raw, &a); err != nil { + continue + } + made = append(made, a) + } + if err := rows.Err(); err != nil { + return nil, err + } + } + seen := map[string]bool{} + var out []string + for _, row := range made { + for _, a := range row { reference := asRecorded(a.Reference) if reference == "" || seen[reference] { continue @@ -369,7 +418,7 @@ func (i *Inventory) everyReferenceMade(ctx context.Context) ([]string, error) { out = append(out, reference) } } - return out, rows.Err() + return out, nil } // asRecorded is an artifact reference in the one vocabulary the sweep speaks (novox/hq issue 226). diff --git a/internal/inventory/fortest.go b/internal/inventory/fortest.go index cbd7f73a..4da06c6b 100644 --- a/internal/inventory/fortest.go +++ b/internal/inventory/fortest.go @@ -12,6 +12,7 @@ import ( "github.com/jackc/pgx/v5" "golang.org/x/crypto/bcrypt" + "github.com/novox/mesh-controller/internal/chain" "github.com/novox/mesh-controller/internal/store" ) @@ -64,6 +65,13 @@ func ForTest(t *testing.T) *Inventory { if _, err := inv.store.Migrate(t.Context(), migrations); err != nil { t.Fatal(err) } + // The chain is on the bus in a mesh; in a test it is in memory, with every kind moved, as a mesh raised after this + // release has them (ADR 0408, issue 1089). The tests of the move itself use ForTestBefore. + inv.UseChain(chain.InMemory(nil)) + if _, err := inv.store.Pool().Exec(t.Context(), + `insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0}`); err != nil { + t.Fatal(err) + } return inv } @@ -121,6 +129,8 @@ func ForTestBefore(t *testing.T, number int) (*Inventory, func()) { if err := inv.Ready(t.Context(), 20*time.Second); err != nil { t.Fatal(err) } + // The chain is in memory and nothing is moved: the tests of a move seed the tables and move them. + inv.UseChain(chain.InMemory(nil)) return inv, func() { t.Helper() if _, err := inv.store.Migrate(t.Context(), all); err != nil { diff --git a/internal/inventory/gate.go b/internal/inventory/gate.go index b5b76768..9932b559 100644 --- a/internal/inventory/gate.go +++ b/internal/inventory/gate.go @@ -187,8 +187,8 @@ func (i *Inventory) PreviousBuild(ctx context.Context, module, commit string, fa } else if bad { continue } - var manifest []byte - if err := i.store.Pool().QueryRow(ctx, `select manifest from build where id = $1`, b.ID).Scan(&manifest); err != nil { + manifest, err := i.manifestOf(ctx, b.ID) + if err != nil { return Build{}, false, err } if len(manifest) == 0 || string(manifest) == "null" { @@ -241,6 +241,11 @@ func (i *Inventory) RestoreModule(ctx context.Context, b Build) error { // machine — its artifacts and its manifest, hashed — so a rebuild that changes nothing there is told // from one that does (the bus's planned step, ADR 0236). func (i *Inventory) BuildFingerprints(ctx context.Context, module string) (map[string]string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainBuildFingerprints(ctx, c, module) + } rows, err := i.store.Pool().Query(ctx, `select commit_hash, made, coalesce(manifest::text, '') from build where module = $1 and failed = '' and commit_hash <> '' order by at desc limit 50`, module) @@ -266,6 +271,11 @@ func (i *Inventory) BuildFingerprints(ctx context.Context, module string) (map[s // Fingerprints is BuildFingerprints for every module at once: module → commit → fingerprint. func (i *Inventory) Fingerprints(ctx context.Context) (map[string]map[string]string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainFingerprints(ctx, c) + } rows, err := i.store.Pool().Query(ctx, `select module, commit_hash, made, coalesce(manifest::text, '') from build where module is not null and failed = '' and commit_hash <> '' order by at desc`) @@ -316,6 +326,11 @@ func (i *Inventory) PassedCommits(ctx context.Context) (map[string]map[string]bo // BuildOf is the id of the newest successful build of a module from a commit; empty when none is // recorded (a manifest handed over by hand). func (i *Inventory) BuildOf(ctx context.Context, module, commit string) (string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return "", err + } else if on { + return i.chainBuildOf(ctx, c, module, commit) + } var id string err := i.store.Pool().QueryRow(ctx, `select id from build where module = $1 and commit_hash = $2 and failed = '' order by at desc limit 1`, @@ -335,19 +350,13 @@ func (i *Inventory) ManifestAt(ctx context.Context, module, commit string) (cata if commit == "" { return catalogue.Manifest{}, false, nil } - var id string - var raw []byte - err := i.store.Pool().QueryRow(ctx, - `select id, manifest from build - where module = $1 and failed = '' and manifest is not null and manifest::text <> 'null' - and (commit_hash = $2 or starts_with(commit_hash, $2)) - order by at desc limit 1`, module, commit).Scan(&id, &raw) - if errors.Is(err, pgx.ErrNoRows) { - return catalogue.Manifest{}, false, nil - } + id, raw, found, err := i.newestManifestAt(ctx, module, commit) if err != nil { return catalogue.Manifest{}, false, err } + if !found { + return catalogue.Manifest{}, false, nil + } if stands, same, err := i.StandingBuild(ctx, module, id); err != nil { return catalogue.Manifest{}, false, err } else if stands != "" && stands != id && len(same) > 0 { @@ -360,3 +369,41 @@ func (i *Inventory) ManifestAt(ctx context.Context, module, commit string) (cata } return m, true, nil } + +// manifestOf is the manifest a build was recorded with; nil when it kept none. +func (i *Inventory) manifestOf(ctx context.Context, id string) ([]byte, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainManifestOf(ctx, c, id) + } + var manifest []byte + if err := i.store.Pool().QueryRow(ctx, `select manifest from build where id = $1`, id).Scan(&manifest); err != nil { + return nil, err + } + return manifest, nil +} + +// newestManifestAt is the newest successful build of a module from a commit, or one it abbreviates, that kept a manifest: +// its id and manifest; false when there is none. +func (i *Inventory) newestManifestAt(ctx context.Context, module, commit string) (string, []byte, bool, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return "", nil, false, err + } else if on { + return i.chainManifestBuild(ctx, c, module, commit) + } + var id string + var raw []byte + err := i.store.Pool().QueryRow(ctx, + `select id, manifest from build + where module = $1 and failed = '' and manifest is not null and manifest::text <> 'null' + and (commit_hash = $2 or starts_with(commit_hash, $2)) + order by at desc limit 1`, module, commit).Scan(&id, &raw) + if errors.Is(err, pgx.ErrNoRows) { + return "", nil, false, nil + } + if err != nil { + return "", nil, false, err + } + return id, raw, true, nil +} diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index 11386553..e422b75a 100644 --- a/internal/inventory/nodes.go +++ b/internal/inventory/nodes.go @@ -18,6 +18,7 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/novox/mesh-controller/internal/broker" + "github.com/novox/mesh-controller/internal/chain" "github.com/novox/mesh-controller/internal/store" ) @@ -32,6 +33,11 @@ type Inventory struct { // configValues reads a module's config values from mesh-config once the move of config has run (ReadsConfigFrom, // issue 664); nil reads none. configValues ConfigValues + // chain is the controller's chain on the bus (novox/hq ADR 0408, issue 1089), once attached (UseChain); moved is which + // kinds of it have been moved from the tables, as chain_moved says (ChainReads). A kind not moved is read and written + // in the tables as before. + chain *chain.Chain + moved *movedKinds } // Open connects to the inventory store. @@ -40,7 +46,7 @@ func Open(ctx context.Context) (*Inventory, error) { if err != nil { return nil, err } - return &Inventory{store: s}, nil + return &Inventory{store: s, moved: &movedKinds{}}, nil } func (i *Inventory) Close() { i.store.Close() } diff --git a/internal/inventory/pending.go b/internal/inventory/pending.go index 0e761e31..8b739e28 100644 --- a/internal/inventory/pending.go +++ b/internal/inventory/pending.go @@ -3,11 +3,14 @@ package inventory import ( "context" "errors" + "fmt" "path" "strings" "time" "github.com/jackc/pgx/v5" + + "github.com/novox/mesh-controller/internal/chain" ) // What the controller asked to be built, and the assignments waiting for it (novox/hq issue 325, ADR 0261). @@ -104,21 +107,36 @@ 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 { - _, err := i.store.Pool().Exec(ctx, - `update build_request set not_asked = $2 - where id = $1 and not exists (select 1 from build where build.id = $1)`, id, why) + if heard, err := i.outcomeHeard(ctx, id); err != nil || heard { + return err + } + _, err := i.store.Pool().Exec(ctx, `update build_request set not_asked = $2 where id = $1`, id, why) return err } // 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 { - _, err := i.store.Pool().Exec(ctx, - `update build_request set outcome_unknown = $2 - where id = $1 and not exists (select 1 from build where build.id = $1)`, id, why) + if heard, err := i.outcomeHeard(ctx, id); err != nil || heard { + return err + } + _, err := i.store.Pool().Exec(ctx, `update build_request set outcome_unknown = $2 where id = $1`, id, why) return err } +// outcomeHeard says a build of this id was recorded: 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 +} + // RequestsNamed is every build request kept whose directory names the module, or whose build said it built // it, newest first, each with its outcome. func (i *Inventory) RequestsNamed(ctx context.Context, name string) ([]RequestOutcome, error) { @@ -154,6 +172,11 @@ func (i *Inventory) RequestedNames(ctx context.Context) ([]string, error) { } func (i *Inventory) requests(ctx context.Context) ([]RequestOutcome, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, err + } else if on { + return i.chainRequests(ctx, c) + } rows, err := i.store.Pool().Query(ctx, `select a.id, a.repository, a.seat, a.source_path, a.ref, a.commit_hash, a.asked_for, coalesce(a.not_asked, ''), coalesce(a.outcome_unknown, ''), a.asked_at, @@ -176,6 +199,45 @@ 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/internal/inventory/source.go b/internal/inventory/source.go index 15504244..63bb9dbc 100644 --- a/internal/inventory/source.go +++ b/internal/inventory/source.go @@ -32,6 +32,11 @@ type sourceRow struct { // sourceRows is every successful build of each module (of one module, when named), newest request // first. func (i *Inventory) sourceRows(ctx context.Context, module string) (map[string][]sourceRow, []string, error) { + if c, on, err := i.ChainReads(ctx, KindBuilds); err != nil { + return nil, nil, err + } else if on { + return i.chainSourceRows(ctx, c, module) + } query := `select b.module, b.id, b.commit_hash, b.source_fingerprint, b.made, b.manifest, coalesce(g.verdict = 'failed', false) from build b left join build_gate g on g.build = b.id diff --git a/testdata/store-tests b/testdata/store-tests index 589272ff..c3536357 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -139,3 +139,9 @@ ./cmd/mesh-controller TestADeliveryGroupMovingAModuleMachinesRunPassesTheGate ./cmd/mesh-controller TestAModuleMovesOnTheOperatorsApproveOfTheMergeThatMovedIt ./cmd/mesh-controller TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheBus +./internal/inventory TestTheMoveCopiesTheBuildRowsAndAgainWritesNothing +./internal/inventory TestAControllerRestartedMidMoveFinishesItAndOverwritesNothing +./internal/inventory TestTheMoveRefusesAValueOverTheLargestMessage +./internal/inventory TestTheMoveRefusesOnACountThatDiffers +./internal/inventory TestTheMoveWithoutTheLeaseWritesNothing +./internal/inventory TestBuildsOnTheBusAreReadBackByEveryReader