diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go index 7a92f89e..adce7906 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go @@ -3,6 +3,7 @@ package main import ( "context" "fmt" + "slices" "time" "github.com/nats-io/nats.go" @@ -22,10 +23,16 @@ type findingsWriter = views.HeldWriter[condseat.Finding, condseat.FindingCleared // runner (design 47 G §266, A-206). const KindPipelineWaits = "pipeline-waits" +// awaitedSaid opens the evidence entry of a pipeline-waits finding that names the awaited runner fingerprint, `awaited ` +// (A-276): the runner clears a finding by that entry alone, never by its summary. +const awaitedSaid = "awaited " + // lateFinding is the finding pipeline-waits of one pipeline the bound's expiry started: scope module, id its key -// `..`, naming the fingerprint of the new runner it waited for and since when. +// `..`, naming the fingerprint of the new runner it waited for and since when, the fingerprint also in one +// evidence entry `awaited ` (A-276). func lateFinding(p seat.Pipeline, held seat.Held, runner, node string, at time.Time) condseat.Finding { return condseat.Finding{Scope: views.FindingScope, ID: p.Pipeline, Kind: KindPipelineWaits, Severity: condseat.FindingSeverityWarning, + Evidence: []condseat.Evidence{{At: at, Said: awaitedSaid + held.Fingerprint}}, Headline: "the new pipeline runner did not run healthy in time", Summary: fmt.Sprintf("the pipeline %s was held since %s for the runner at %s to run healthy; the bound passed and the running runner runs it", p.Pipeline, held.Since.UTC().Format(time.RFC3339), held.Fingerprint), @@ -51,3 +58,33 @@ func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, j } return w, nil } + +// heldFindings is what clearAwaited needs of the runner's held finding writer: what it holds and a clear. +type heldFindings interface { + Held() []condseat.Finding + Clear(ctx context.Context, scope, id, kind, why string) error +} + +// clearAwaited clears exactly the pipeline-waits findings whose evidence names fp awaited (A-218, A-276): the node-engine seat's +// running says the runner at fp runs healthy. The writer holds what its state findings hold from its start, so a finding raised +// before a restart is cleared too. It says how many it cleared, and the first clear that failed. +func clearAwaited(ctx context.Context, w heldFindings, fp string) (int, error) { + if w == nil || fp == "" { + return 0, nil + } + n := 0 + var first error + for _, f := range w.Held() { + if f.Kind != KindPipelineWaits || !slices.ContainsFunc(f.Evidence, func(e condseat.Evidence) bool { return e.Said == awaitedSaid+fp }) { + continue + } + if err := w.Clear(ctx, f.Scope, f.ID, f.Kind, "the runner at "+fp+" runs healthy"); err != nil { + if first == nil { + first = err + } + continue + } + n++ + } + return n, first +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_integration_test.go new file mode 100644 index 00000000..d4936c40 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_integration_test.go @@ -0,0 +1,110 @@ +//go:build integration + +package main + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// Design 47 G A-218, A-276: a pipeline-waits finding raised before the runner restarted is held again by the new process's writer, +// seeded from the state findings, and is cleared when the node-engine seat's running says the awaited runner fingerprint healthy; +// the finding awaiting another fingerprint stands. +func TestAPipelineWaitsFindingIsClearedAfterARestartWhenItsAwaitedRunnerRunsHealthy(t *testing.T) { + url := testbus.URL(t) + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + const module = "mesh-delivery-pipeline" + states := map[string]jetstream.KeyValue{} + for _, state := range []string{"findings", "pipelines", "bounds"} { + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}) + if err != nil { + t.Fatal(err) + } + states[state] = kv + } + // Where the writer says its findings, as the bus keeps the module's events. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.>"}}); err != nil { + t.Fatal(err) + } + cred := process.Account{URL: url, Node: "n1", Module: module, User: "runner", Password: "x"} + conn, err := process.Connect(cred) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + since := time.Now().UTC().Add(-time.Hour) + before, err := findingWriter(ctx, cred, nc, js, conn.State("findings"), t.Logf) + if err != nil { + t.Fatal(err) + } + for _, p := range []struct{ key, module, fp string }{{"main.4.app", "app", "src2:new"}, {"main.5.web", "web", "src2:newer"}} { + if err := before.Write(ctx, lateFinding(seat.Pipeline{Pipeline: p.key, Module: p.module, Branch: "main"}, + seat.Held{Since: since, Fingerprint: p.fp}, module, "n1", since.Add(30*time.Minute))); err != nil { + t.Fatal(err) + } + } + before.Close() + keys := func() map[string]bool { + out := map[string]bool{} + ks, err := states["findings"].Keys(ctx) + if err != nil && err != jetstream.ErrNoKeysFound { + t.Fatal(err) + } + for _, k := range ks { + out[k] = true + } + return out + } + app, web := "module.main.4.app.pipeline-waits", "module.main.5.web.pipeline-waits" + if k := keys(); !k[app] || !k[web] { + t.Fatalf("the state findings holds %v before the restart, want %s and %s", k, app, web) + } + + // The restart: a new process's writer holds what its state findings hold. + after, err := findingWriter(ctx, cred, nc, js, conn.State("findings"), t.Logf) + if err != nil { + t.Fatal(err) + } + defer after.Close() + if len(after.Held()) != 2 { + t.Fatalf("the writer after the restart holds %+v, want both findings", after.Held()) + } + hub := &oneHub{} + starter := &pipeline.Starter{Logf: t.Logf} + if err := hearRunning(ctx, hub, module, starter, pipelineStore{ctx: ctx, kv: states["pipelines"]}, states["bounds"], after, + func(string) (pipeline.Half, bool) { return pipeline.Half{}, false }, t.Logf); err != nil { + t.Fatal(err) + } + healthy := true + raw, _ := json.Marshal(nodeengine.Running{V: 1, Node: "n1", Module: module, Fingerprint: "src2:new", Healthy: &healthy, At: time.Now().UTC()}) + var acks []string + hub.handle(dataMsg{data: raw, acks: &acks}) + if k := keys(); k[app] || !k[web] { + t.Errorf("after the runner at src2:new runs healthy the state findings holds %v: want %s cleared and %s standing", k, app, web) + } + if held := after.Held(); len(held) != 1 || held[0].ID != "main.5.web" { + t.Errorf("the writer holds %+v, want only the finding awaiting src2:newer", held) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go index 8d9dd3fc..b6ec439b 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go @@ -1,6 +1,8 @@ package main import ( + "context" + "slices" "strings" "testing" "time" @@ -26,4 +28,30 @@ func TestLateFindingIsPipelineWaitsOfTheModuleScope(t *testing.T) { if err := views.CheckFinding(f); err != nil { t.Error(err) } + // A-276: the awaited fingerprint is in one evidence entry, `awaited `. + if len(f.Evidence) != 1 || f.Evidence[0].Said != "awaited src2:new" { + t.Errorf("evidence %+v, want one entry `awaited src2:new`", f.Evidence) + } + // A-218, A-276: a healthy running of the runner at a fingerprint clears exactly the pipeline-waits findings whose evidence + // names it awaited: not one awaiting another, not another kind naming it, never by the summary. + other := lateFinding(seat.Pipeline{Pipeline: "main.5.web", Module: "web"}, seat.Held{Since: since, Fingerprint: "src2:newer"}, "r", "n1", since) + other.Summary = f.Summary // a summary naming src2:new is never read + kind := f + kind.Kind, kind.ID = "step-failed", "main.4.app.build" + held := &memFindings{held: []condseat.Finding{f, other, kind}} + if n, err := clearAwaited(context.Background(), held, "src2:new"); err != nil || n != 1 || !slices.Equal(held.cleared, []string{"module.main.4.app.pipeline-waits"}) { + t.Errorf("cleared %d (%v): %v, want the one pipeline-waits awaiting src2:new", n, err, held.cleared) + } +} + +// memFindings is a held finding writer's Held and Clear. +type memFindings struct { + held []condseat.Finding + cleared []string +} + +func (m *memFindings) Held() []condseat.Finding { return m.held } +func (m *memFindings) Clear(_ context.Context, scope, id, kind, _ string) error { + m.cleared = append(m.cleared, scope+"."+id+"."+kind) + return nil } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 024d800d..78a1e312 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -229,7 +229,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } hearBuildEnds(ctx, cred, js, ends, starter, store, logger.Printf) - if err := hearRunning(ctx, heard, cred.Module, starter, store, bounds, rv.Catalog.View, logger.Printf); err != nil { + if err := hearRunning(ctx, heard, cred.Module, starter, store, bounds, rv.Findings, rv.Catalog.View, logger.Printf); err != nil { return err } if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index 6b263676..813e01a5 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -192,10 +192,11 @@ func armHeld(ctx context.Context, bounds jetstream.KeyValue) func(string, time.T } // hearRunning releases the pipelines held for the runner's new version when the node-engine seat says the runner's own module -// runs that fingerprint healthy, on any node (§266, §276, A-195); its bound is then deleted. A running with no health reads -// as healthy (§520). Heard again when the release could not be written. +// runs that fingerprint healthy, on any node (§266, §276, A-195); its bound is then deleted, and the pipeline-waits findings +// that await it are cleared (A-218, A-276). A running with no health reads as healthy (§520). Heard again when the release or +// a clear could not be written. func hearRunning(ctx context.Context, hub handlers, self string, st *pipeline.Starter, store pipeline.Updater, bounds jetstream.KeyValue, - view func(string) (pipeline.Half, bool), logf func(string, ...any)) error { + findings heldFindings, view func(string) (pipeline.Half, bool), logf func(string, ...any)) error { return hub.Handle(nodeengine.SubjectRunningAll, func(m jetstream.Msg) { var ev nodeengine.Running if err := json.Unmarshal(m.Data(), &ev); err != nil { @@ -219,6 +220,15 @@ func hearRunning(ctx context.Context, hub handlers, self string, st *pipeline.St logf("the bound of the wait for %s is not deleted: %v", ev.Fingerprint, err) } } + cleared, err := clearAwaited(ctx, findings, ev.Fingerprint) + if cleared > 0 { + logf("the runner runs %s on %s: %d pipeline-waits findings cleared", ev.Fingerprint, ev.Node, cleared) + } + if err != nil { + logf("a pipeline-waits finding awaiting %s is not cleared: %v; heard again", ev.Fingerprint, err) + _ = m.Nak() + return + } _ = m.Ack() }) }