diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go index b730238d..d184ad30 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/findings.go @@ -2,38 +2,41 @@ package main import ( "context" + "encoding/json" "errors" + "fmt" + "strings" "time" - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" - "git.novox.be/novox/mesh-sdk/go/process" - + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-builder/internal/versions" ) // findingsBucket is the builder's findings (novox/hq ADR 0564 §2): what it sees wrong now, written at its raise and // deleted at its clear, which the condition keeper raises and clears. -var findingsBucket = meshconditionsclient.FindingsBucket(builds.BuilderModule) +var findingsBucket = builds.BuilderModule + "_" + mcseat.StateFindings -// findings writes the builder's findings into its state, once bound; until then each is said in the log only. -type findings struct { - kv *process.AwaitedBucket - logf func(string, ...any) - now func() time.Time - // store stands in for kv in a test. - store findingStore +// heldWriter is the builder's finding writer: its findings are raised as they happen, so the writer holds them (design 47 G +// A-29); its `findings-asked` subject is the seat's, from the generated contract (A-17). +type heldWriter = views.HeldWriter[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent] + +// findingWriter is the part of the held writer the builder's findings use. +type findingWriter interface { + Write(ctx context.Context, f mcseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error } -// findingStore is the part of the findings bucket written. -type findingStore interface { - Get(ctx context.Context, key string) (jetstream.KeyValueEntry, error) - Put(ctx context.Context, key string, value []byte) (uint64, error) - Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error +// findings writes the builder's findings through its held finding writer (design 47 G §689, A-29): a finding written before the +// writer starts is held and stated at its start. +type findings struct { + w findingWriter + logf func(string, ...any) + now func() time.Time } // finding is one of the builder's findings: its kind, what it is about, and its words. @@ -41,21 +44,43 @@ type finding struct { kind, scope, id string summary, headline, explanation, resolved, needs string severity string + machine, said, source string } -func (f *findings) bucket() findingStore { - switch { - case f == nil: - return nil - case f.store != nil: - return f.store - case f.kv == nil: - return nil +// seatFinding is fd as the seat's generated finding at at, its id made safe as a key's part. +func (fd finding) seatFinding(at time.Time) mcseat.Finding { + at = at.UTC() + id := keyID(fd.id) + sev := fd.severity + if sev == "" { + sev = mcseat.FindingSeverityWarning } - if kv := f.kv.KV(); kv != nil { - return kv + said, source := fd.said, fd.source + if said == "" { + said = fd.summary } - return nil + if source == "" { + source = "builder." + fd.kind + } + return mcseat.Finding{Scope: fd.scope, ID: id, Kind: fd.kind, Severity: sev, Summary: fd.summary, Headline: fd.headline, + Explanation: fd.explanation, Needs: fd.needs, Source: source, Observer: builds.BuilderModule, At: at, + Evidence: []mcseat.Evidence{{At: at, Said: said}}, + Subject: mcseat.Subject{Scope: fd.scope, ID: id, Machine: fd.machine}} +} + +// keyID is an id as the parts of a key: each dot-separated part made safe for a bus key, as the key of a condition is. +func keyID(id string) string { + key := mcc.Key("x", id, "x") + return key[len("x.") : len(key)-len(".x")] +} + +// callUnanswered is the finding of a kept call to callee about id, unanswered since since, its last fault said (ADR 0668 §3): +// its key is `call..call-unanswered`. +func callUnanswered(id, callee, machine, fault, source string, since time.Time) finding { + return finding{scope: mcc.ScopeCall, id: id, kind: mcc.KindCallUnanswered, machine: machine, said: fault, source: source, + summary: "the call " + callee + " about " + id + " is unanswered since " + since.UTC().Format(time.RFC3339), + headline: "A call whose answer is kept is not answered", + explanation: "The call " + callee + " is made again on every transport fault and on the callee's event; it has not been answered since " + since.UTC().Format(time.RFC3339) + ". Its last fault: " + fault + "."} } // BuildFailed is the finding of a failed trunk build (DLV-B7, design 25), about its module. @@ -69,106 +94,63 @@ func (f *findings) buildEnded(ctx context.Context, e builds.BuildLogEntry) { } switch e.Outcome { case builds.BuildFailed: - f.raise(ctx, finding{kind: BuildFailed, scope: meshconditionsclient.ScopeModule, id: e.Module, + f.raise(ctx, finding{kind: BuildFailed, scope: mcc.ScopeModule, id: e.Module, summary: "the trunk build of " + e.Module + " at " + e.FP + " (commit " + e.Commit + ") failed", headline: "A module's trunk build failed", explanation: "The builder built " + e.Module + " from the trunk and the build failed, so its new version is not delivered.", resolved: "A later trunk build of " + e.Module + " is done.", needs: "builder_builds_since shows the build; builder_build_again builds it again."}) case builds.BuildDone: - f.clear(ctx, BuildFailed, meshconditionsclient.ScopeModule, e.Module) + f.clear(ctx, BuildFailed, mcc.ScopeModule, e.Module) } } -// raise writes fd; it is written again only when its words changed, so the keeper's condition moves only then. +// raise writes fd through the writer. func (f *findings) raise(ctx context.Context, fd finding) { - f.raiseObservation(ctx, meshconditionsclient.Observation{Scope: fd.scope, ID: fd.id, Kind: fd.kind, Severity: fd.severity, - Summary: fd.summary, Headline: fd.headline, Explanation: fd.explanation, Resolved: fd.resolved, Needs: fd.needs, - Source: "builder." + fd.kind}) -} - -// raiseObservation writes o, as raise does. -func (f *findings) raiseObservation(ctx context.Context, o meshconditionsclient.Observation) { - kv := f.bucket() - if kv == nil { - f.logf("finding %s (not written, %s is not on the bus yet): %s", o.Key(), findingsBucket, o.Summary) + x := fd.seatFinding(f.now()) + if f.w == nil { + f.logf("finding %s (not written, no finding writer): %s", views.FindingKey(x.Scope, x.ID, x.Kind), x.Summary) return } - if _, err := meshconditionsclient.Raise(findingBucket{ctx, kv}, builds.BuilderModule, "", o, f.now()); err != nil { - f.logf("the finding %s cannot be written: %v", o.Key(), err) + if err := f.w.Write(ctx, x); err != nil { + f.logf("the finding %s cannot be written: %v", views.FindingKey(x.Scope, x.ID, x.Kind), err) } } -// clear deletes the finding of kind about scope and id; one not there is nothing to clear. +// clear clears the finding of kind about scope and id; one not there is nothing to clear. func (f *findings) clear(ctx context.Context, kind, scope, id string) { - kv := f.bucket() - if kv == nil { + if f.w == nil { return } - if _, err := meshconditionsclient.Clear(findingBucket{ctx, kv}, scope, id, kind); err != nil { - f.logf("the finding %s cannot be cleared: %v", meshconditionsclient.Key(scope, id, kind), err) + if err := f.w.Clear(ctx, scope, keyID(id), kind, "the builder no longer observes it"); err != nil { + f.logf("the finding %s cannot be cleared: %v", views.FindingKey(scope, keyID(id), kind), err) } } -// findingBucket is the builder's findings store as the bucket the conditions client raises and clears in. -type findingBucket struct { - ctx context.Context - kv findingStore -} - -func (b findingBucket) Get(key string) ([]byte, bool, error) { - e, err := b.kv.Get(b.ctx, key) - switch { - case errors.Is(err, jetstream.ErrKeyNotFound): - return nil, false, nil - case err != nil: - return nil, false, err - } - return e.Value(), true, nil -} - -func (b findingBucket) Put(key string, value []byte) error { - _, err := b.kv.Put(b.ctx, key, value) - return err -} - -func (b findingBucket) Delete(key string) error { - if err := b.kv.Delete(b.ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { - return err - } - return nil -} - -// notHeard is the writer of the builder's finding of a family it does not hear (heard.Writer, ADR 0615 §5): written into -// its findings once bound, said in the log until then. +// notHeard is the writer of the builder's finding of a family it does not hear (heard.Writer, ADR 0615 §5): written through the +// finding writer, said in the log until it starts. func (f *findings) notHeard(ctx context.Context) heard.Writer { - on := func(do func(findingStore)) { - switch { - case f.store != nil: - do(f.store) - case f.kv != nil: - f.kv.Ready(func(kv jetstream.KeyValue) { do(kv) }) - } - } return heard.Writer{Observer: builds.BuilderModule, Logf: f.logf, Now: f.now, - Put: func(key string, value []byte) error { - on(func(kv findingStore) { - if _, err := kv.Put(ctx, key, value); err != nil { - f.logf("the finding %s cannot be written: %v", key, err) - } - }) - return nil + Put: func(_ string, value []byte) error { + var x mcseat.Finding + if err := json.Unmarshal(value, &x); err != nil { + return err + } + if f.w == nil { + return errors.New("no finding writer") + } + return f.w.Write(ctx, x) }, Delete: func(key string) error { - on(func(kv findingStore) { - if _, err := kv.Get(ctx, key); err != nil { - return // none stands - } - if err := kv.Delete(ctx, key); err != nil { - f.logf("the finding %s cannot be cleared: %v", key, err) - } - }) - return nil + scope, rest, _ := strings.Cut(key, ".") + i := strings.LastIndex(rest, ".") + if i < 0 { + return fmt.Errorf("%q is no finding's key", key) + } + if f.w == nil { + return nil + } + return f.w.Clear(ctx, scope, rest[:i], rest[i+1:], "the builder hears it") }} } @@ -177,10 +159,10 @@ func (f *findings) notHeard(ctx context.Context) heard.Writer { func (f *findings) floorSaid(ctx context.Context) func(versions.FloorWait, bool) { return func(w versions.FloorWait, raised bool) { if !raised { - f.clear(ctx, versions.FloorSilent, meshconditionsclient.ScopeModule, w.Name) + f.clear(ctx, versions.FloorSilent, mcc.ScopeModule, w.Name) return } - f.raise(ctx, finding{kind: versions.FloorSilent, scope: meshconditionsclient.ScopeModule, id: w.Name, summary: w.Finding(), + f.raise(ctx, finding{kind: versions.FloorSilent, scope: mcc.ScopeModule, id: w.Name, summary: w.Finding(), headline: "A version's floor is not known", explanation: "The builder numbers a name only once its registry seat, the forge bridge's tags and the indexer's requirements all answered; one of them did not.", resolved: "All three answered.", diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index fbfc0f50..7feaa294 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -309,6 +309,11 @@ func startBedWith(t *testing.T, closed bool) *bed { AllowDirect: true}); err != nil { t.Fatal(err) } + // The builder's own stream of its findings' events (design 47 G A-30, §383), which its finding writer says into. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "BUILDER_FINDINGS", Subjects: []string{"mesh.stream.mesh-delivery-builder.event.finding-written.*", + "mesh.stream.mesh-delivery-builder.event.finding-cleared.*", "mesh.stream.mesh-delivery-builder.event.findings-stated.*"}}); err != nil { + t.Fatal(err) + } b.fake = &fakeBuildModules{js: js, ctx: ctx, calls: map[string]int{}} b.fake.closed.Store(closed) b.fake.serve(t, "mesh-build-go", "n1", "n2") 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 fa35f4ac..b873b4c2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -21,7 +21,7 @@ import ( "time" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" - sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" @@ -92,7 +92,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro hub := process.NewEventHub(logger.Printf) reader := &views.ProcessReader{Module: builds.BuilderModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} findingsKV := process.AwaitBucket(ctx, js, conn, findingsBucket, "the builder's findings, a first view unanswered past its bound among them, are said in its log only", logger.Printf) - reader.Findings = func() views.FindingWriter { return sdkfindings.Writer(ctx, findingsKV.KV) } + // The builder's findings are raised as they happen, so its one finding writer holds them (design 47 G A-29, §689): it + // is started once the state is bound, and what is written before is stated at its start. + fw, err := views.NewHeldWriter(views.WriterConfig[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent]{ + Module: cred.Module, State: calls.State("findings"), Node: cred.Node, Asked: mcseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logger.Printf}) + if err != nil { + return err + } + findingsKV.Ready(func(jetstream.KeyValue) { + if err := fw.Start(ctx); err != nil { + logger.Printf("the finding writer cannot start, its findings are kept in the state alone: %v", err) + } + }) + reader.Findings = func() views.WireWriter { return fw } modules, err := reader.Follow(indexerModule, modulesState, indexerModule+".indexer_read_modules") if err != nil { return err @@ -151,7 +163,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The builder numbers each trunk build at its claim (novox/hq design 51 *The versions*): its state versions, bound once // the bus holds it; the floor's three sources asked as its own account. versionsKV := process.AwaitBucket(ctx, js, conn, versions.Bucket, "no trunk build is numbered: each is built unnumbered and published by its digest", logger.Printf) - found := &findings{kv: findingsKV, logf: logger.Printf, now: time.Now} + found := &findings{w: fw, logf: logger.Printf, now: time.Now} floorWaits := &versions.FloorWaits{Logf: logger.Printf, Said: found.floorSaid(context.WithoutCancel(ctx))} // The names waiting on their floor are kept in the state floor-waits (DLV-B4): a restarted builder takes them in, so // their silent sources' events still wake them; until the state is bound they are kept here only. diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go index 4217ca6b..2cd14157 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go @@ -43,9 +43,9 @@ func heardBy(call func(key string, body any) (json.RawMessage, error)) func(cont func (r *engineReads) callsOf(f *findings, bounds func() jetstream.KeyValue, heard func(context.Context, string) (bool, error), logf func(string, ...any)) *enginereads.ReadsCalls { c := &enginereads.ReadsCalls{Keeper: r.keeper, Nodes: r.named, Heard: heard, Bound: mcc.CallUnansweredAfter, Logf: logf, Raise: func(ctx context.Context, u enginereads.UnansweredCall) error { - o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "builder."+mcc.KindCallUnanswered, u.Since) - o.Needs = "builder_engine_reads with ask_again asks the node again now and shows its statement." - f.raiseObservation(ctx, o) + o := callUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "builder."+mcc.KindCallUnanswered, u.Since) + o.needs = "builder_engine_reads with ask_again asks the node again now and shows its statement." + f.raise(ctx, o) return nil }, Clear: func(ctx context.Context, node string) error { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index cafc3e8a..af0276d6 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -15,7 +15,9 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -151,7 +153,7 @@ func theCountedNodesEngineReads(t *testing.T) { asked = append(asked, key, body) return json.Marshal(meshwatchdogs.PresenceAnswer{Machine: "n2", Online: online}) }) - calls := r2.callsOf(&findings{store: st, logf: t.Logf, now: clock}, func() jetstream.KeyValue { return nil }, heard, t.Logf) + calls := r2.callsOf(&findings{w: st, logf: t.Logf, now: clock}, func() jetstream.KeyValue { return nil }, heard, t.Logf) calls.Arm = func(_ context.Context, node string, ends time.Time) error { armed[node] = ends; return nil } key := meshconditionsclient.Key(meshconditionsclient.ScopeCall, "manifest-reads.n2", meshconditionsclient.KindCallUnanswered) r2.placed() // n2 asked, unanswered at 1000 @@ -165,7 +167,7 @@ func theCountedNodesEngineReads(t *testing.T) { } online = true r2.expiredCall("n2") - f, err := meshconditionsclient.DecodeFinding(st.vals[key]) + f, err := decodeFinding(st.vals[key]) if err != nil || f.Observer != builds.BuilderModule || f.Kind != meshconditionsclient.KindCallUnanswered || len(asked) != 4 { t.Fatalf("heard five minutes on: finding %+v, %v (keys %v), presence asked %v", f, err, slices.Collect(maps.Keys(st.vals)), asked) } @@ -327,7 +329,7 @@ func (failingVersions) KeysOf(context.Context, string) ([]string, error) { func aFailedTrunkBuildIsAFinding(t *testing.T) { st := &memFindings{vals: map[string][]byte{}} - f := &findings{store: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} + f := &findings{w: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} ended := func(outcome, ref string) builds.BuildLogEntry { return builds.BuildLogEntry{Event: builds.BuildEndedEvent, Module: "m", FP: "f1", Commit: "c", Ref: ref, Outcome: outcome} } @@ -337,7 +339,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { t.Fatalf("a candidate's failed build is a finding: %v", st.vals) } f.buildEnded(context.Background(), ended(builds.BuildFailed, "")) - got, err := meshconditionsclient.DecodeFinding(st.vals[key]) + got, err := decodeFinding(st.vals[key]) if err != nil || got.Observer != builds.BuilderModule || !strings.Contains(got.Summary, "trunk build of m at f1") { t.Fatalf("the finding = %+v, %v", got, err) } @@ -349,7 +351,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { fw := &versions.FloorWaits{Said: f.floorSaid(context.Background())} fw.Wait("m/lib", versions.SeatGoPackages, "", versions.Floor{Silent: []string{versions.SourceTags}}, nil, time.Unix(0, 0)) floorKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, "m/lib", versions.FloorSilent) - if got, err := meshconditionsclient.DecodeFinding(st.vals[floorKey]); err != nil || !strings.Contains(got.Summary, "bridge_tags did not answer") { + if got, err := decodeFinding(st.vals[floorKey]); err != nil || !strings.Contains(got.Summary, "bridge_tags did not answer") { t.Fatalf("the floor finding = %+v, %v", got, err) } fw.Answered("m/lib") @@ -374,23 +376,23 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { if !h.heard(context.Background(), refused) || !refused.acked { t.Fatalf("a refusal: acked %v", refused.acked) } - if got, err := meshconditionsclient.DecodeFinding(st.vals[refusedKey]); err != nil || !strings.Contains(got.Summary, "refused to publish m/program@0.1.3") || !strings.Contains(got.Summary, "bytes differ") { + if got, err := decodeFinding(st.vals[refusedKey]); err != nil || !strings.Contains(got.Summary, "refused to publish m/program@0.1.3") || !strings.Contains(got.Summary, "bytes differ") { t.Fatalf("a holder's refused: the finding = %+v, %v", got, err) } pubs.ended(context.Background(), purged{key: versions.PublishBoundKey("m/program", "0.1.4")}) unansweredKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, "m/program", versions.PublishUnanswered) - if got, err := meshconditionsclient.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "neither published nor refused of m/program@0.1.4") { + if got, err := decodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "neither published nor refused of m/program@0.1.4") { t.Fatalf("a bound ended unanswered: the finding = %+v, %v", got, err) } // Review round 2: a bound that cannot be written is publish-unanswered at once; a watch of bounds that cannot start // is family-not-heard naming the state. delete(st.vals, unansweredKey) pubs.arm(context.Background(), versions.Entry{Name: "m/program", Version: "0.1.4", Seat: "oci-artifact-registry"}) - if got, err := meshconditionsclient.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "its bound could not be written") { + if got, err := decodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "its bound could not be written") { t.Fatalf("a bound not written: the finding = %+v, %v", got, err) } watchBounds(context.Background(), refusedWatch{err: errors.New("permissions violation")}, func(jetstream.KeyValueEntry) {}, f.notHeard(context.Background()), nil, t.Logf) - if got, err := meshconditionsclient.DecodeFinding(st.vals[heard.Key(builds.BuilderModule, versions.BoundsState)]); err != nil || !strings.Contains(got.Summary, "permissions violation") { + if got, err := decodeFinding(st.vals[heard.Key(builds.BuilderModule, versions.BoundsState)]); err != nil || !strings.Contains(got.Summary, "permissions violation") { t.Fatalf("bounds not watched: the finding = %+v, %v", got, err) } delete(st.vals, heard.Key(builds.BuilderModule, versions.BoundsState)) @@ -432,7 +434,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { } keys.memVersions[buildKey], looking = building, false u.expired(context.Background(), purged{key: versions.UnputBoundKey(buildKey)}) - if got, err := meshconditionsclient.DecodeFinding(st.vals[callKey]); err != nil || got.Kind != meshconditionsclient.KindCallUnanswered || !strings.Contains(got.Summary, buildKey) || !strings.Contains(got.Needs, build.BuildAgainTool) { + if got, err := decodeFinding(st.vals[callKey]); err != nil || got.Kind != meshconditionsclient.KindCallUnanswered || !strings.Contains(got.Summary, buildKey) || !strings.Contains(got.Needs, build.BuildAgainTool) { t.Fatalf("the bound's end of a key left building: the finding = %+v, %v", got, err) } u.ended(context.Background(), buildKey) @@ -441,7 +443,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { } u.bounds = func() jetstream.KeyValue { return nil } u.left(context.Background(), buildKey, "nats: timeout") - if got, err := meshconditionsclient.DecodeFinding(st.vals[callKey]); err != nil || !strings.Contains(got.Explanation, "nor its bound") { + if got, err := decodeFinding(st.vals[callKey]); err != nil || !strings.Contains(got.Explanation, "nor its bound") { t.Fatalf("a bound not written: the finding = %+v, %v", got, err) } delete(st.vals, callKey) @@ -449,7 +451,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { notHeard := f.notHeard(context.Background()) unheardKey := heard.Key(builds.BuilderModule, "FORGE") _ = notHeard.Bind("FORGE", "a branch's end is not heard", func() error { return errors.New("consumer not found") }) - if got, err := meshconditionsclient.DecodeFinding(st.vals[unheardKey]); err != nil || got.Kind != heard.Kind || !strings.Contains(got.Summary, "does not hear FORGE") { + if got, err := decodeFinding(st.vals[unheardKey]); err != nil || got.Kind != heard.Kind || !strings.Contains(got.Summary, "does not hear FORGE") { t.Fatalf("a consumer not there: the finding = %+v, %v", got, err) } _ = notHeard.Bind("FORGE", "a branch's end is not heard", func() error { return nil }) @@ -499,7 +501,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) { } now = now.Add(versions.FindingAfter) window.step(context.Background(), []string{"a", "b"}, counted) - if got, err := meshconditionsclient.DecodeFinding(st.vals[heldKey]); err != nil || !strings.Contains(got.Summary, "the node-engine of b") { + if got, err := decodeFinding(st.vals[heldKey]); err != nil || !strings.Contains(got.Summary, "the node-engine of b") { t.Fatalf("seven days held: the finding = %+v, %v", got, err) } counted["b"] = newest @@ -581,25 +583,21 @@ func (p purged) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePurg // memFindings is a findings bucket in memory. type memFindings struct{ vals map[string][]byte } -type memEntry struct { - jetstream.KeyValueEntry - v []byte +// Write keeps the finding under its key, as the held writer does in the state. +func (m *memFindings) Write(_ context.Context, f mcseat.Finding) error { + v, err := json.Marshal(f) + m.vals[views.FindingKey(f.Scope, f.ID, f.Kind)] = v + return err } -func (e memEntry) Value() []byte { return e.v } - -func (m *memFindings) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { - v, ok := m.vals[key] - if !ok { - return nil, jetstream.ErrKeyNotFound - } - return memEntry{v: v}, nil -} -func (m *memFindings) Put(_ context.Context, key string, v []byte) (uint64, error) { - m.vals[key] = v - return 1, nil -} -func (m *memFindings) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { - delete(m.vals, key) +// Clear deletes the finding of scope, id and kind. +func (m *memFindings) Clear(_ context.Context, scope, id, kind, _ string) error { + delete(m.vals, views.FindingKey(scope, id, kind)) return nil } + +// decodeFinding reads a finding as the writer kept it. +func decodeFinding(v []byte) (mcseat.Finding, error) { + var f mcseat.Finding + return f, json.Unmarshal(v, &f) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go index 679e1b95..dca519a1 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go @@ -75,7 +75,7 @@ func (b *boundsKV) Delete(_ context.Context, key string, _ ...jetstream.KVDelete func TestTheCatalogsWordsMoveTheReservations(t *testing.T) { ctx := context.Background() st := &memFindings{vals: map[string][]byte{}} - found := &findings{store: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} + found := &findings{w: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} bounds := &boundsKV{keys: map[string]bool{}} woke := 0 call := func(addr string, body any) (json.RawMessage, error) { @@ -109,7 +109,7 @@ func TestTheCatalogsWordsMoveTheReservations(t *testing.T) { t.Fatal("a reservation's bound was not taken in") } key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, "m", versions.VersionUnreleased) - got, err := meshconditionsclient.DecodeFinding(st.vals[key]) + got, err := decodeFinding(st.vals[key]) if err != nil || !strings.Contains(got.Summary, "m:b unpublished") || strings.Contains(got.Summary, "m:a unpublished") || !strings.Contains(got.Summary, "1.4.3") { t.Fatalf("finding = %+v (%v), want version-unreleased naming m:b of 1.4.3", got, err) } @@ -146,7 +146,7 @@ func TestAReleasedVersionIsTaggedAtTheCommitItWasBuiltFrom(t *testing.T) { tags = append(tags, body.(git.TagInput)) return json.Marshal(git.TagAnswer{Performed: refuse == "", Error: refuse}) } - found := &findings{store: &memFindings{vals: map[string][]byte{}}, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} + found := &findings{w: &memFindings{vals: map[string][]byte{}}, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }} rv := newReservations(&memReserve{vals: map[string][]byte{}, revs: map[string]uint64{}}, call, func() jetstream.KeyValue { return &boundsKV{keys: map[string]bool{}} }, found, nil, t.Logf) rv.filled.Store(true) goPkg := versions.Source{Repository: "o/r", Commit: "c0ffee", GoPackage: true} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go index 55c9ccab..19817b41 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go @@ -78,7 +78,7 @@ func (u *unputs) expired(ctx context.Context, e jetstream.KeyValueEntry) { } func (u *unputs) raise(ctx context.Context, key, fault string, since time.Time) { - o := meshconditionsclient.CallUnanswered(key, unputCallee, "", fault, "builder."+meshconditionsclient.KindCallUnanswered, since) - o.Needs = build.BuildAgainTool + " builds the key again: an ask its stream holds is waited for, one never put is put; " + build.StuckTool + " shows it." - u.findings.raiseObservation(ctx, o) + o := callUnanswered(key, unputCallee, "", fault, "builder."+meshconditionsclient.KindCallUnanswered, since) + o.needs = build.BuildAgainTool + " builds the key again: an ask its stream holds is waited for, one never put is put; " + build.StuckTool + " shows it." + u.findings.raise(ctx, o) } diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index b35b278e..79789582 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -155,7 +155,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if want := []string{"container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed", "go-package-registry.kept", "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed", "mesh-artifact-sweep.artifacts-gone", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled", - "mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.assignments", "mesh-delivery-forge-bridge.forge.**", + "mesh-conditions.condition-cleared", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", "mesh-delivery-forge-bridge.forge.**", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.packages", "node-build-go.toolchain-reported", "node-build-node.toolchain-reported", "node-engine.manifest-reads-changed", "node-engine.running", "npm-package-registry.kept", "npm-package-registry.published", "npm-package-registry.refused", "npm-package-registry.removed", @@ -174,8 +174,10 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 t.Errorf("tools = %v, want %v", m.Tools, want) } // Design 47 G §672: the toolchain waits' stream, kept 7 days within 16 MiB, and the bound's setting, 60 minutes. - if len(m.Streams) != 1 || m.Streams[0].Name != "waits" || !reflect.DeepEqual(m.Streams[0].Events, []string{"wait-held.**", "wait-ended.**"}) || m.Streams[0].MaxAgeDays != 7 || m.Streams[0].MaxMiB != 16 { - t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB", m.Streams) + // Design 47 G A-30, §383: and its own findings' events, which its finding writer says. + if len(m.Streams) != 2 || m.Streams[0].Name != "waits" || !reflect.DeepEqual(m.Streams[0].Events, []string{"wait-held.**", "wait-ended.**"}) || m.Streams[0].MaxAgeDays != 7 || m.Streams[0].MaxMiB != 16 || + m.Streams[1].Name != "mesh-delivery-builder-findings" || !reflect.DeepEqual(m.Streams[1].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) { + t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB, and the findings' stream", m.Streams) } if c, ok := m.Config["toolchain-wait-minutes"]; !ok || c.Default != float64(60) || len(m.Config) != 1 { t.Errorf("config = %+v, want toolchain-wait-minutes of 60", m.Config) diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 0e428436..ac053db9 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -40,6 +40,7 @@ "version.old-form-closed" ], "consumes": [ + "mesh-conditions.findings-asked", "container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", @@ -116,7 +117,8 @@ { "name": "first-views", "history": 1 } ], "streams": [ - { "name": "waits", "events": ["wait-held.**", "wait-ended.**"], "max-age-days": 7, "max-mib": 16 } + { "name": "waits", "events": ["wait-held.**", "wait-ended.**"], "max-age-days": 7, "max-mib": 16 }, + { "name": "mesh-delivery-builder-findings", "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], "max-age-days": 7, "max-mib": 16 } ], "config": { "toolchain-wait-minutes": {