From b44f01f89ce734988bb66229d7aca855b2e5f2cd Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 23:13:32 +0100 Subject: [PATCH] =?UTF-8?q?Builder:=20write=20each=20build's=20start=20and?= =?UTF-8?q?=20end=20to=20its=20own=20log=20on=20the=20bus,=20and=20ask=20t?= =?UTF-8?q?he=20seat=20artifact-store's=20held=20and=20hear=20its=20artifa?= =?UTF-8?q?cts-gone,=20the=20controller's=20only=20until=20the=20lease=20w?= =?UTF-8?q?rite,=20so=20build=20history=20and=20the=20store's=20sweep=20le?= =?UTF-8?q?ave=20the=20controller=20(issue=201955,=20ADR=200571=20=C2=A71,?= =?UTF-8?q?=20=C2=A72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pins mesh-sdk f9a496d. --- .../cmd/mesh-delivery-builder/main.go | 8 ++++ .../cmd/mesh-delivery-builder/reasks.go | 23 +++++++++ .../cmd/mesh-delivery-builder/stored.go | 16 ++++++- modules/mesh-delivery-builder/go.mod | 2 +- modules/mesh-delivery-builder/go.sum | 4 +- .../internal/build/run.go | 48 +++++++++++++++++++ .../internal/build/run_test.go | 19 ++++++++ .../mesh-delivery-builder/manifest_test.go | 26 +++++++--- modules/mesh-delivery-builder/module.json | 13 ++++- 9 files changed, 146 insertions(+), 13 deletions(-) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 7fb207c..13b7dcb 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -120,6 +120,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Manifests: manifestReader{manifests}, Logf: logger.Printf, Stored: storedVia(calls.Call), + Node: cred.Node, } // A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base // revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision @@ -129,6 +130,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro trunks := &trunkSource{ctx: ctx, js: js, heads: &sourceHeads{}, moved: b.TrunkMoved, errs: errs, logf: logger.Printf} trunks.start() b.Trunk = trunks.head + // The builder's log (ADR 0571 §1): each build's start and end, on its own stream, read by the view, the indexer, the + // assigner and mesh-catalog. + b.Log = func(e delivery.BuildLogEntry) { + if err := emitLog(context.WithoutCancel(ctx), js, cred, e); err != nil { + logger.Printf("%v", err) + } + } b.Said = func(e build.AskedAgain) { if err := emitAskedAgain(context.WithoutCancel(ctx), js, cred, e); err != nil { logger.Printf("%v", err) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go index 311ba12..def0bd1 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go @@ -8,6 +8,7 @@ import ( "strings" "time" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -138,3 +139,25 @@ func announcedAgain(ctx context.Context, b askAgainer, kv jetstream.KeyValue, e logf("%s.%s: the answer could not be written: %v", build.ReAsksState, e.Key(), err) } } + +// emitLog says one entry of the builder's log on its stream `log` (ADR 0571 §1), under an id the stream keeps one of. +func emitLog(ctx context.Context, js jetstream.JetStream, cred process.Account, e delivery.BuildLogEntry) error { + data, err := delivery.EncodeBuildLogEntry(e) + if err != nil { + return err + } + msg := nats.NewMsg(process.ModuleStreamSubject(cred.Module, e.Event)) + id := e.Event + "." + e.Module + "." + e.FP + "." + e.At.Format(time.RFC3339Nano) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", cred.Module) + msg.Header.Set("x-node", cred.Node) + msg.Header.Set("x-time", e.At.Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + pub, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)); err != nil { + return fmt.Errorf("logging %s of %s at %s: %w", e.Event, e.Module, e.FP, err) + } + return nil +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go index 3f2bbef..fab1825 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go @@ -9,7 +9,9 @@ import ( "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" "mesh-delivery-builder/internal/build" ) @@ -18,17 +20,27 @@ import ( // consumer of artifacts-gone (mesh-controller internal/broker EventsStream). const eventsStream = "EVENTS" +// SeatHeld is the seat artifact-store's verb held (ADR 0571 §2). +const SeatHeld = "seat:" + catalogue.ArtifactStoreSeat + ".held" + // toolCaller calls a tool or a seat verb as the builder's own account (mesh-sdk process.Conn.Call). type toolCaller func(key string, body any) (json.RawMessage, error) // storedVia is build.Builder's Stored: it asks the controller's verb held (mesh-sdk delivery.HeldVerb; novox/hq issue // 1937) and answers the references the store no longer holds. One it could not ask about, with none gone, is an error: // the key is then left done, and the controller's take-in still refuses a gone blob. +// +// It asks the seat artifact-store's `held` (novox/hq ADR 0571 §2), and the controller's while that seat has no holder that +// answers (a named exception of clause 10, until the lease write). func storedVia(call toolCaller) func(ctx context.Context, references []string) ([]string, error) { return func(_ context.Context, references []string) ([]string, error) { - raw, err := call(delivery.HeldVerb, map[string]any{"references": references}) + body := map[string]any{"references": references} + raw, err := call(SeatHeld, body) + if handacts.NoResponder(err) { + raw, err = call(delivery.HeldVerb, body) + } if err != nil { - return nil, fmt.Errorf("asking %s: %w", delivery.HeldVerb, err) + return nil, fmt.Errorf("asking held: %w", err) } a, err := delivery.DecodeHeldAnswer(raw, references) if err != nil { diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 1031a9e..38e5c36 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-builder go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-builder/go.sum b/modules/mesh-delivery-builder/go.sum index 043e05f..ecec070 100644 --- a/modules/mesh-delivery-builder/go.sum +++ b/modules/mesh-delivery-builder/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a h1:nhRFmtdcI4VkghGUdJhpt1BnPuTQcekioOIRzW8fcG0= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029200316-22c9ddd32b6a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 935860b..aeb84b2 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -7,6 +7,7 @@ import ( "fmt" "maps" "sort" + "strconv" "strings" "sync" "time" @@ -68,6 +69,11 @@ type Builder struct { Askings int // Said is told each ask asked again (the builder's event build.asked-again). Nil tells nobody. Said func(AskedAgain) + // Log is told each build's start and end, the entries of the builder's log (novox/hq ADR 0571 §1). Nil tells nobody. + Log func(delivery.BuildLogEntry) + // Now is the clock of the log; nil is time.Now. Node is the node the builder runs on, which asked the build. + Now func() time.Time + Node string once sync.Once sem chan struct{} @@ -429,6 +435,7 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { return } b.handClaimed(key, cl.Rev) + b.logBuild(LogStarted(module, m, b.Node, b.now())) if cl.Unasked != nil { b.logf("%v", cl.Unasked) } @@ -478,6 +485,7 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) { b.logf("%s: the build could not be written: %v", cl.Key, err) return } + b.logBuild(LogEnded(module, m, b.Node, b.now(), result)) if result.State == delivery.BuildFailed { b.logf("%s: failed: %s", cl.Key, result.Error) } @@ -872,3 +880,43 @@ func needsFetch(tasks []Task) bool { } return false } + +func (b *Builder) now() time.Time { + if b.Now != nil { + return b.Now() + } + return time.Now() +} + +// logBuild tells the log an entry, said in the journal when it cannot be one. +func (b *Builder) logBuild(e delivery.BuildLogEntry) { + if b.Log == nil { + return + } + if _, err := delivery.EncodeBuildLogEntry(e); err != nil { + b.logf("%s at %s: no log entry: %v", e.Module, e.FP, err) + return + } + b.Log(e) +} + +// LogStarted is the log's entry of a build of module's fingerprint starting (ADR 0571 §1): the module, the fingerprint, the +// source revision, where it is built from (repository, directory, commit), the node of the builder that asked it and the time. +func LogStarted(module string, m delivery.Module, by string, at time.Time) delivery.BuildLogEntry { + dir := m.Path + if dir == "" { + dir = "." + } + return delivery.BuildLogEntry{Event: delivery.BuildStartedEvent, Module: module, FP: m.FP2, Source: strconv.FormatUint(m.Src, 10), + Repository: m.Repository, Directory: dir, Commit: m.Commit, Node: by, At: at.UTC()} +} + +// LogEnded is the log's entry of that build ending: the outcome (done, failed or queued) and each artifact's digest. +func LogEnded(module string, m delivery.Module, by string, at time.Time, result delivery.BuilderBuild) delivery.BuildLogEntry { + e := LogStarted(module, m, by, at) + e.Event, e.Outcome = delivery.BuildEndedEvent, result.State + for _, a := range result.Artifacts { + e.Artifacts = append(e.Artifacts, delivery.BuiltDigest{Name: a.Name, Digest: a.Digest}) + } + return e +} diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index 6cbb41c..9a4a10f 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -221,7 +221,26 @@ var ( func TestABundleAndAnArchiveAreBuiltOnOneNodeAndDone(t *testing.T) { // 1607/AC7, ADR 0478 clauses 1, 4 r := newRig(t, goBundle, archiveA, tsBundle) + var logged []delivery.BuildLogEntry + var lmu sync.Mutex + r.b.Node, r.b.Log = "novox", func(e delivery.BuildLogEntry) { lmu.Lock(); logged = append(logged, e); lmu.Unlock() } v := r.consider(t, "modules/m") + // 1955/L7 (ADR 0571 §1): the build is one start and one end in the builder's log, every field named, its duration read + // from the two. + if len(logged) != 2 || logged[0].Event != delivery.BuildStartedEvent || logged[1].Event != delivery.BuildEndedEvent || + logged[1].Outcome != delivery.BuildDone || len(logged[1].Artifacts) != 3 || logged[0].Repository != "novox/mesh-catalog" || + logged[0].Directory != "modules/m" || logged[0].Commit != "7a74cee83f88e6585a5511c3cc53bdf0f5303d20" || logged[0].Node != "novox" || + logged[0].FP != fp1 || logged[0].Module != "m" { + t.Fatalf("logged = %+v", logged) + } + for _, e := range logged { + if _, err := delivery.EncodeBuildLogEntry(e); err != nil { + t.Errorf("an entry the log refuses: %v", err) + } + } + if _, ok := delivery.BuildDuration(logged[0], logged[1]); !ok { + t.Error("no duration from the two entries") + } if v.State != delivery.BuildDone || v.By != "novox.builder" || v.Manifest != manifestD || len(v.Artifacts) != 3 { t.Fatalf("build = %+v, want done by novox.builder with three artifacts", v) } diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index cf0cac7..a43e1f9 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -40,10 +40,15 @@ type manifest struct { Emits []string `json:"emits"` Consumes []string `json:"consumes"` State []state `json:"state"` - Reads []string `json:"reads"` - Uses []string `json:"uses"` - Invokes []string `json:"invokes"` - Config map[string]struct { + Streams []struct { + Name string `json:"name"` + Events []string `json:"events"` + MaxAgeDays int `json:"max-age-days"` + } `json:"streams"` + Reads []string `json:"reads"` + Uses []string `json:"uses"` + Invokes []string `json:"invokes"` + Config map[string]struct { Type string `json:"type"` Description string `json:"description"` Pattern string `json:"pattern"` @@ -101,12 +106,14 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 } // 1937/AC3: the builder asks the controller whether the artifact store still holds a done key's blobs (held), and no // other tool; it holds no capacity. - if want := []string{"seat:mesh-controller.held"}; !reflect.DeepEqual(m.Invokes, want) || len(m.Config) != 0 { + // 1955/L7 (ADR 0571 §2): the seat artifact-store's held, the controller's only while the seat has no holder (clause 10). + if want := []string{"seat:artifact-store.held", "seat:mesh-controller.held"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want) } // The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it // starts; and by the controller's artifacts-gone, a take-in refused for a gone blob (1937/AC4). - if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-controller.artifacts-gone"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + // 1955/L7: and the seat artifact-store's, the controller's until the lease write (its take-in sweeps until then). + if want := []string{"artifact-store.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-controller.artifacts-gone"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -122,7 +129,12 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) { t.Errorf("emits = %v, want %v", m.Emits, want) } - for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} { + // 1955/L7 (ADR 0571 §1): its log, each build's start and end, kept 30 days. + if len(m.Streams) != 1 || m.Streams[0].Name != "log" || m.Streams[0].MaxAgeDays != 30 || + !reflect.DeepEqual(m.Streams[0].Events, []string{"build.started", "build.ended"}) { + t.Errorf("streams = %+v, want the log", m.Streams) + } + for _, k := range []string{"logs", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the builder has none", k) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index faf5694..98c1b0b 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -13,11 +13,13 @@ "build.asked-again" ], "consumes": [ + "artifact-store.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-controller.artifacts-gone" ], "invokes": [ + "seat:artifact-store.held", "seat:mesh-controller.held" ], "state": [ @@ -27,6 +29,16 @@ { "name": "queues", "history": 1 }, { "name": "reasks", "history": 1 } ], + "streams": [ + { + "name": "log", + "events": [ + "build.started", + "build.ended" + ], + "max-age-days": 30 + } + ], "reads": [ "mesh-build-go.steps", "mesh-build-go.toolchains", @@ -44,7 +56,6 @@ "node-build-image", "node-build-node" ], - "own-secrets": { "broker": "${dir:state}/broker" },