G findings4 builder: the builder's findings are the seat's generated finding written through views.NewHeldWriter, its findings stream in manifest and bed
Builds: design 47 G A-29, A-30, A-17, A-26, A-3, §689, §686, §383 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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.<id>.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.",
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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": {
|
||||
|
||||
Reference in New Issue
Block a user