diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drops.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drops.go new file mode 100644 index 00000000..dc1fde33 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drops.go @@ -0,0 +1,92 @@ +package main + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go/jetstream" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// callFindings is the runner's finding writer as its drops use it (design 47 G A-222, §350). +type callFindings interface { + Write(ctx context.Context, f condseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + +// dropFindingID is the id of the call-unanswered finding of buildSeat's drop of ask. +func dropFindingID(buildSeat, ask string) string { return buildSeat + ".drop." + ask } + +// deleteBound deletes a key of the state bounds; a key already gone is no fault. +func deleteBound(ctx context.Context, bounds jetstream.KeyValue) func(key string) error { + return func(key string) error { + if err := bounds.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil + } +} + +// armDropBound arms the five-minute bound of a drop, a key of the state bounds with its own time to live (A-77, A-222). +func armDropBound(ctx context.Context, bounds jetstream.KeyValue) func(key string) error { + return func(key string) error { + _, err := bounds.Create(ctx, key, []byte(`{"drop":"`+key+`"}`), jetstream.KeyTTL(pipeline.AskBound)) + if errors.Is(err, jetstream.ErrKeyExists) { + return nil + } + return err + } +} + +// dropAnswered ends the wait of an answered drop: its bound is deleted and a call-unanswered finding it raised before is +// cleared. +func dropAnswered(ctx context.Context, unbound func(key string) error, w callFindings) func(buildSeat, ask string) error { + return func(buildSeat, ask string) error { + if err := unbound(pipeline.DropBoundKey(buildSeat, ask)); err != nil { + return err + } + return w.Clear(ctx, mcc.ScopeCall, dropFindingID(buildSeat, ask), mcc.KindCallUnanswered, "the drop is answered") + } +} + +// dropUnanswered says the drop of ask on buildSeat unanswered, the finding call-unanswered in the runner's findings, on the +// expiry of its five-minute bound (A-222, §350, ADR 0668 §3). +func dropUnanswered(ctx context.Context, w callFindings, module string, now func() time.Time) func(buildSeat, ask string) error { + return func(buildSeat, ask string) error { + if w == nil { + return errors.New("the runner has no finding writer") + } + id := dropFindingID(buildSeat, ask) + call := buildSeat + ".drop" + return w.Write(ctx, condseat.Finding{Scope: mcc.ScopeCall, ID: id, Kind: mcc.KindCallUnanswered, Severity: condseat.FindingSeverityWarning, + Summary: fmt.Sprintf("the call %s of ask %s is unanswered", call, ask), + Headline: "A call whose answer is kept is not answered", + Explanation: fmt.Sprintf("The drop of the superseded step's ask %s was asked of the seat %s as a kept call and no holder answered it within %s, so the build it names may still run.", ask, buildSeat, pipeline.AskBound), + Source: module + "." + mcc.KindCallUnanswered, Observer: module, At: now().UTC(), Subject: condseat.Subject{Scope: mcc.ScopeCall, ID: id}}) + } +} + +// boundExpired takes the expiry of a key of the state bounds: a drop's says the drop unanswered, an ask's ends the step no +// holder took (§32, §535, A-77); another key is none of the runner's. +func boundExpired(h *pipeline.Hearer, unanswered func(buildSeat, ask string) error, logf func(string, ...any)) func(key string) { + return func(key string) { + if s, ask, ok := pipeline.ParseDropBound(key); ok { + if err := unanswered(s, ask); err != nil { + logf("the drop of ask %s on %s, unanswered, is not said: %v", ask, s, err) + } + return + } + if !pipeline.IsAskKey(key) { + return + } + if err := h.NotTaken(key, pipeline.AskBound); err != nil { + logf("the step of ask %s, which no holder took, is not ended: %v", key, err) + } + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drops_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drops_test.go new file mode 100644 index 00000000..8b8321b0 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/drops_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "context" + "testing" + "time" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + + "mesh-delivery-pipeline/internal/pipeline" +) + +type fakeFindings struct { + written []condseat.Finding + cleared []string +} + +func (f *fakeFindings) Write(_ context.Context, x condseat.Finding) error { + f.written = append(f.written, x) + return nil +} + +func (f *fakeFindings) Clear(_ context.Context, scope, id, kind, _ string) error { + f.cleared = append(f.cleared, scope+"."+id+"."+kind) + return nil +} + +// A-222, A-77: the expiry of a drop's five-minute bound says the drop call-unanswered in the runner's findings, naming the +// seat and the ask; an answered drop clears it; a key that is no drop's is not said. +func TestADropBoundsExpirySaysCallUnanswered(t *testing.T) { + w := &fakeFindings{} + var logged int + expired := boundExpired(nil, dropUnanswered(context.Background(), w, "the-runner", func() time.Time { return time.Unix(100, 0) }), func(string, ...any) { logged++ }) + expired("not-a-bound") + if len(w.written) != 0 { + t.Fatalf("a key of no drop was said: %+v", w.written) + } + expired(pipeline.DropBoundKey("node-build-go", "step-ab")) + if len(w.written) != 1 { + t.Fatalf("written %+v, want one finding", w.written) + } + f := w.written[0] + if f.Scope != mcc.ScopeCall || f.Kind != mcc.KindCallUnanswered || f.ID != "node-build-go.drop.step-ab" || f.Observer != "the-runner" || f.Subject.ID != f.ID { + t.Errorf("the finding = %+v", f) + } + var unbound []string + if err := dropAnswered(context.Background(), func(k string) error { unbound = append(unbound, k); return nil }, w)("node-build-go", "step-ab"); err != nil || + len(w.cleared) != 1 || w.cleared[0] != "call.node-build-go.drop.step-ab.call-unanswered" || len(unbound) != 1 || unbound[0] != "drop.node-build-go.step-ab" { + t.Errorf("an answered drop: %v, cleared %v, unbound %v", err, w.cleared, unbound) + } +} 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 8a388b9d..0016c85e 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -14,6 +14,7 @@ import ( "strings" "sync/atomic" "syscall" + "time" "github.com/nats-io/nats.go" @@ -160,7 +161,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31, // §350; A-182, A-202). starter.Supersede = func(newer seat.Pipeline) { - n, err := hearer.Supersede(newer, dropAsk(calls, cred.Module), dropBound(ctx, bounds)) + n, err := hearer.Supersede(newer, pipeline.BoundDrop(dropAsk(calls, cred.Module), armDropBound(ctx, bounds), dropAnswered(ctx, deleteBound(ctx, bounds), rv.Findings)), dropBound(ctx, bounds)) if err != nil { logger.Printf("the pipelines %s supersedes: %d ended, then %v", newer.Pipeline, n, err) } else if n > 0 { @@ -176,7 +177,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearAskTaken(ctx, heard, bounds, hearer.Taken, logger.Printf); err != nil { return err } - if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil { + if err := watchBounds(ctx, bounds, boundExpired(hearer, dropUnanswered(ctx, rv.Findings, cred.Module, time.Now), logger.Printf), 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 c75823dc..45d3759d 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -179,8 +179,8 @@ func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm { } // watchBounds hears the end of each ask's bound, the bus's marker of the key's expiry, and ends the step no holder took (§32, -// §535, A-77): no timer in the process. -func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, logf func(string, ...any)) error { +// §535, A-77), and says a drop no holder answered unanswered (A-222): no timer in the process. +func watchBounds(ctx context.Context, bounds jetstream.KeyValue, expired func(key string), logf func(string, ...any)) error { w, err := bounds.WatchAll(ctx, jetstream.UpdatesOnly()) if err != nil { return err @@ -195,12 +195,10 @@ func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hea if !ok { return } - if e == nil || e.Operation() != jetstream.KeyValuePurge || !pipeline.IsAskKey(e.Key()) { + if e == nil || e.Operation() != jetstream.KeyValuePurge { continue } - if err := h.NotTaken(e.Key(), pipeline.AskBound); err != nil { - logf("the step of ask %s, which no holder took, is not ended: %v", e.Key(), err) - } + expired(e.Key()) } } }() diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index eca36ef1..19a132c2 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -20,6 +20,8 @@ import ( type runnerViews struct { Registrations *meshcatalogclient.Registrations Catalog *pipeline.Views + // Findings is the runner's finding writer, the reader's and the runner's own findings through it. + Findings callFindings } // startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or @@ -59,7 +61,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr return nil, err } reader.Start(ctx, js, cred, status) - return &runnerViews{Registrations: regs, Catalog: cv}, nil + return &runnerViews{Registrations: regs, Catalog: cv, Findings: findings}, nil } // catalogView is a branch's meshcatalogclient.Catalog as the runner reads it. diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 9bfc48db..3fcef96d 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -10,6 +10,7 @@ require ( require ( git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go index c4a107b2..44089fcb 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go @@ -93,3 +93,36 @@ func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error { // IsAskKey says whether a key of the state `bounds` is an ask's bound. func IsAskKey(key string) bool { return strings.HasPrefix(key, "step-") && !strings.Contains(key, ".") } + +// DropBoundPrefix begins the key of a superseded step's drop in the state `bounds`, `drop..` (design 47 G +// A-222, A-77): its five-minute expiry says the drop unanswered. +const DropBoundPrefix = "drop." + +// DropBoundKey is the key of the bound of buildSeat's drop of ask. +func DropBoundKey(buildSeat, ask string) string { return DropBoundPrefix + buildSeat + "." + ask } + +// ParseDropBound reads a key of the state `bounds` as a drop's bound: the build seat and the ask it names. +func ParseDropBound(key string) (buildSeat, ask string, ok bool) { + rest, found := strings.CutPrefix(key, DropBoundPrefix) + if !found { + return "", "", false + } + buildSeat, ask, found = strings.Cut(rest, ".") + return buildSeat, ask, found && buildSeat != "" && ask != "" && !strings.Contains(ask, ".") +} + +// BoundDrop is drop with a bound of five minutes (A-222, A-77): the bound key is armed, with its own time to live, before +// the drop is asked, and answered is told when the drop is answered, to delete the key, so an answer ends the wait with no +// finding; a drop never answered leaves its key to expire, and the expiry says it unanswered. A bound that cannot be armed +// is returned and the drop not asked. +func BoundDrop(drop Drop, arm func(key string) error, answered func(buildSeat, ask string) error) Drop { + return func(buildSeat, node, ask, why string) error { + if err := arm(DropBoundKey(buildSeat, ask)); err != nil { + return fmt.Errorf("the bound of the drop of ask %s is not armed: %w", ask, err) + } + if err := drop(buildSeat, node, ask, why); err != nil { + return err + } + return answered(buildSeat, ask) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go index d6b96b54..8490be86 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/supersede_test.go @@ -1,6 +1,7 @@ package pipeline import ( + "errors" "slices" "strings" "testing" @@ -91,3 +92,34 @@ func TestSupersede(t *testing.T) { t.Errorf("the older publish's pipeline is %s: %q", p.Outcome, p.Why) } } + +// A-222, A-77: a drop is bounded by a key of its own armed before it is asked and deleted when it is answered; one never +// answered leaves its key to expire. +func TestBoundDrop(t *testing.T) { + var armed, answered []string + arm := func(key string) error { armed = append(armed, key); return nil } + done := func(s, ask string) error { answered = append(answered, s+"/"+ask); return nil } + if s, ask, ok := ParseDropBound(DropBoundKey(nodebuildgo.Seat, "step-ab")); !ok || s != nodebuildgo.Seat || ask != "step-ab" { + t.Errorf("the drop's key reads %q, %q, %v", s, ask, ok) + } + if _, _, ok := ParseDropBound("step-ab"); ok { + t.Error("an ask's bound was read as a drop's") + } + if IsAskKey(DropBoundKey("s", "step-ab")) { + t.Error("a drop's bound was taken for an ask's") + } + err := BoundDrop(func(string, string, string, string) error { return nil }, arm, done)(nodebuildgo.Seat, "n1", "step-ab", "why") + if err != nil || len(armed) != 1 || armed[0] != "drop."+nodebuildgo.Seat+".step-ab" || len(answered) != 1 { + t.Errorf("an answered drop: %v, armed %v, answered %v", err, armed, answered) + } + answered = nil + err = BoundDrop(func(string, string, string, string) error { return errors.New("no responders") }, arm, done)(nodebuildgo.Seat, "", "step-cd", "why") + if err == nil || len(armed) != 2 || len(answered) != 0 { + t.Errorf("an unanswered drop: %v, armed %v, answered %v (its key is left to expire)", err, armed, answered) + } + asked := false + err = BoundDrop(func(string, string, string, string) error { asked = true; return nil }, func(string) error { return errors.New("full") }, done)(nodebuildgo.Seat, "", "step-ef", "why") + if err == nil || asked { + t.Errorf("a drop whose bound is not armed: %v, asked %v", err, asked) + } +}