From 4238318601364393dff8ba7df0590fbad618f273 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:54:07 +0200 Subject: [PATCH] Pin mesh-sdk at feat/r4g's head a9560890, re-vendor it and capture its conformance events there (R4g) --- go.mod | 2 +- go.sum | 4 +- testdata/beside/CAPTURED | 2 +- .../novox/mesh-sdk/go/buildstep/pull.go | 9 + .../novox/mesh-sdk/go/buildstep/pull_fake.go | 8 +- .../mesh-sdk/go/delivery/buildrun/asks.go | 187 +++++++++++------- .../novox/mesh-sdk/go/delivery/checker.go | 9 +- .../novox/mesh-sdk/go/delivery/stepasks.go | 11 ++ vendor/modules.txt | 2 +- 9 files changed, 154 insertions(+), 80 deletions(-) diff --git a/go.mod b/go.mod index 993261e5..06a9d5c0 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-controller go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023141925-9205222a98bb + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023145022-a9560890ee26 github.com/jackc/pgx/v5 v5.10.0 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 diff --git a/go.sum b/go.sum index 2697a741..ff18345f 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,7 @@ git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYgA9TCv+4K8JgaURg3zQppvUgJLaDMP4= git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023141925-9205222a98bb h1:D70wIl0aHcEXhb9/7aDV0uDrP+A5wLsBcQ5kEad7QTQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023141925-9205222a98bb/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023145022-a9560890ee26 h1:onj4dRQ/fZ+jA68ICqMgWOzTzbS4X1lj0If/GO0f1oc= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023145022-a9560890ee26/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/testdata/beside/CAPTURED b/testdata/beside/CAPTURED index 9685c506..808135da 100644 --- a/testdata/beside/CAPTURED +++ b/testdata/beside/CAPTURED @@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself. mesh-catalog 2246ccc492e59579a140752e6f574b2121d296af modules/*/module.json, modules/*/files, modules/*/templates, predefined.json mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock -mesh-sdk 9205222a98bb0d460e3a827d69d2ff640c500b47 conformance/events/module-event.json +mesh-sdk a9560890ee26edca8b538be8429a041dfb105654 conformance/events/module-event.json To move them, from this repository's root, with the two repositories checked out beside it: diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull.go b/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull.go index 32f243e9..9fb676b3 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull.go @@ -281,6 +281,15 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { _ = msg.Nak() return } + if msg.Delivered > delivery.AskRuns { + // Run AskRuns times and never ended: each run took its node down or died with it. It is run no more, and said + // (issue 1907): the asker's key fails with why. + why := fmt.Errorf("the ask was taken %d times and no run of it ended: each took its build node down or died with it, so it is run no more (poison)", msg.Delivered-1) + p.logf("the ask %s is ended on %s: %v", ask.ID, p.Node, why) + finish(p.Put, key, step, nil, why) + _ = msg.Term() + return + } module := "" if ask.Build != nil { module = ask.Build.Module diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull_fake.go b/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull_fake.go index 2e6624dd..1e10e4af 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull_fake.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull_fake.go @@ -17,6 +17,8 @@ type FakeAsks struct { // Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given. Acked, Termed, Naked []string InProgress int + // Delivered, when set, is the delivery count every ask handed over says (1 when 0): a test of a poisoned ask sets it. + Delivered uint64 } // NewFakeAsks is an empty double. @@ -102,6 +104,10 @@ func (f *FakeAsks) pulled(data []byte) *Pulled { return nil } } - return &Pulled{Data: data, Delivered: 1, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), + delivered := f.Delivered + if delivered == 0 { + delivered = 1 + } + return &Pulled{Data: data, Delivered: delivered, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }} } diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/buildrun/asks.go b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/buildrun/asks.go index de6dac25..59d99d5f 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/buildrun/asks.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/buildrun/asks.go @@ -22,17 +22,19 @@ type AskPublisher interface { PublishAsk(ctx context.Context, subject, id string, data []byte) error } -// AskWaiter waits for an ask's step: the first key `*.` of the build module's steps state whose state is not -// running. A watch, not a poll; it returns when the step ends or ctx does. +// AskWaiter watches an ask's steps: every value a key `*.` of the build module's steps state takes, running or +// ended, on the channel it answers, until ctx ends. A watch, not a poll. type AskWaiter interface { - WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) + WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) } // Asker puts asks and waits for their steps. type Asker struct { Publisher AskPublisher Waiter AskWaiter - // Bound is how long the asker waits for an ask's step before it stops waiting; zero waits as long as ctx. + // Bound is how long a step may run once a build node took the ask (its key written running) before the asker stops + // waiting; zero waits as long as ctx. An ask waiting in its stream is not bounded: the stream's backlog is no fault + // (ADR 0525 §6, §9), and the bound starts again when another node takes the ask. Bound time.Duration } @@ -53,25 +55,57 @@ func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error { } // Wait waits for the step of an ask already put (a builder that started again takes over its own asks with it, never -// putting them twice) and answers it when it ended, whatever its state. +// putting them twice) and answers it when it ended, whatever its state. The bound runs from the step's start on the node +// that took the ask, read from its running key, so an ask that waits in the stream for a free slot waits as long as ctx. func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) { module, err := delivery.BuildModuleOf(a.Language) if err != nil { return delivery.Step{}, err } - wait, cancel := ctx, context.CancelFunc(func() {}) - if k.Bound > 0 { - wait, cancel = context.WithTimeoutCause(ctx, k.Bound, fmt.Errorf("the ask %s's step did not end within %s", a.ID, k.Bound)) - } + watch, cancel := context.WithCancel(ctx) defer cancel() - step, err := k.Waiter.WaitAsk(wait, module, a.ID) + steps, err := k.Waiter.WatchAsk(watch, module, a.ID) if err != nil { - if cause := context.Cause(wait); cause != nil && ctx.Err() == nil { - err = cause - } return delivery.Step{}, err } - return step, nil + var bound <-chan time.Time + var timer *time.Timer + defer func() { + if timer != nil { + timer.Stop() + } + }() + running, since := "", time.Time{} + for { + select { + case <-ctx.Done(): + return delivery.Step{}, ctx.Err() + case <-bound: + return delivery.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start", a.ID, running, k.Bound) + case step, ok := <-steps: + if !ok { + return delivery.Step{}, errors.New("the watch of the ask's step ended") + } + if step.State != delivery.StepRunning { + return step, nil + } + if k.Bound <= 0 || (step.Node == running && step.Started.Equal(since)) { + continue + } + // A node took the ask, or took it again after its run stopped (another node, or the same node's build module + // replaced mid-run: a new start): its step is bounded from its own start. + running, since = step.Node, step.Started + left := k.Bound + if !step.Started.IsZero() { + left = time.Until(step.Started.Add(k.Bound)) + } + if timer != nil { + timer.Stop() + } + timer = time.NewTimer(max(left, 0)) + bound = timer.C + } + } } // Ask puts a and waits for its step; a step that did not pass is an error naming its node, its state and its reason. @@ -107,53 +141,61 @@ func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data return err } -// WaitAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at +// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at // once, and every change after. -func (s JetStreamAsks) WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) { +func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) { pattern, err := delivery.AskStepWatch(askID) if err != nil { - return delivery.Step{}, err + return nil, err } kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module)) if err != nil { - return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) } w, err := kv.Watch(ctx, pattern) if err != nil { - return delivery.Step{}, err + return nil, err } - defer w.Stop() - for { - select { - case <-ctx.Done(): - return delivery.Step{}, ctx.Err() - case e, ok := <-w.Updates(): - if !ok { - return delivery.Step{}, errors.New("the watch of the ask's step ended") - } - if e == nil || e.Operation() != jetstream.KeyValuePut { - continue - } - step, err := delivery.DecodeStep(e.Value()) - if err != nil { - continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's) - } - if step.State != delivery.StepRunning { - return step, nil + out := make(chan delivery.Step) + go func() { + defer close(out) + defer w.Stop() + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + step, err := delivery.DecodeStep(e.Value()) + if err != nil { + continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's) + } + select { + case out <- step: + case <-ctx.Done(): + return + } } } - } + }() + return out, nil } // FakeAsks is a test double of the bus for an asker's tests: each ask put is kept and, when Answer is set, its step is -// ended at once with what Answer says a build node wrote. Safe for concurrent use. +// written at once with what Answer says a build node wrote (running or ended). Write writes a step later. Safe for +// concurrent use. type FakeAsks struct { Answer func(a delivery.StepAsk) delivery.Step - mu sync.Mutex - asks []delivery.StepAsk - steps map[string]delivery.Step - waits map[string]chan struct{} + mu sync.Mutex + asks []delivery.StepAsk + steps map[string][]delivery.Step + watchers map[string][]chan delivery.Step } // PublishAsk is AskPublisher's. @@ -166,50 +208,53 @@ func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) erro f.asks = append(f.asks, a) f.mu.Unlock() if f.Answer != nil { - f.End(f.Answer(a)) + f.Write(f.Answer(a)) } return nil } -// End writes an ask's step end, as a build node does. -func (f *FakeAsks) End(step delivery.Step) { +// Write writes an ask's step, as a build node does. +func (f *FakeAsks) Write(step delivery.Step) { f.mu.Lock() if f.steps == nil { - f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{} + f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{} } - f.steps[step.ID] = step - ch := f.waits[step.ID] - delete(f.waits, step.ID) + f.steps[step.ID] = append(f.steps[step.ID], step) + ws := append([]chan delivery.Step(nil), f.watchers[step.ID]...) f.mu.Unlock() - if ch != nil { - close(ch) + for _, w := range ws { + w <- step } } -// WaitAsk is AskWaiter's. -func (f *FakeAsks) WaitAsk(ctx context.Context, _ string, askID string) (delivery.Step, error) { +// End is Write: kept for the tests that end a step. +func (f *FakeAsks) End(step delivery.Step) { f.Write(step) } + +// WatchAsk is AskWaiter's: what was written so far, then every write. +func (f *FakeAsks) WatchAsk(ctx context.Context, _ string, askID string) (<-chan delivery.Step, error) { + out := make(chan delivery.Step, 64) f.mu.Lock() if f.steps == nil { - f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{} + f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{} } - if s, ok := f.steps[askID]; ok { - f.mu.Unlock() - return s, nil - } - ch := f.waits[askID] - if ch == nil { - ch = make(chan struct{}) - f.waits[askID] = ch + for _, s := range f.steps[askID] { + out <- s } + f.watchers[askID] = append(f.watchers[askID], out) f.mu.Unlock() - select { - case <-ctx.Done(): - return delivery.Step{}, ctx.Err() - case <-ch: - } - f.mu.Lock() - defer f.mu.Unlock() - return f.steps[askID], nil + go func() { + <-ctx.Done() + f.mu.Lock() + ws := f.watchers[askID] + for i, w := range ws { + if w == out { + f.watchers[askID] = append(ws[:i], ws[i+1:]...) + break + } + } + f.mu.Unlock() + }() + return out, nil } // Asks are the asks put so far. diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/checker.go b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/checker.go index 679985b3..8b7a95c1 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/checker.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/checker.go @@ -120,9 +120,12 @@ type Composition struct { // Took is the time of each part of a check in integer milliseconds (ADR 0447 clause 3): waiting in the queue, cloning, // the unit level, the integration level's setup and its running, and the composition. type Took struct { - Queue int64 `json:"queue"` - Clone int64 `json:"clone"` - Unit int64 `json:"unit"` + Queue int64 `json:"queue"` + Clone int64 `json:"clone"` + Unit int64 `json:"unit"` + // Lint is the unit level's gofmt and go vet before its tests (ADR 0529 §1; issue 1887 D7-go); absent from a verdict + // of a build module that runs none. + Lint int64 `json:"lint,omitempty"` IntegrationSetup int64 `json:"integration-setup"` IntegrationRun int64 `json:"integration-run"` Composition int64 `json:"composition"` diff --git a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stepasks.go b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stepasks.go index a7ae9d21..4e1f001e 100644 --- a/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stepasks.go +++ b/vendor/git.novox.be/novox/mesh-sdk/go/delivery/stepasks.go @@ -41,6 +41,17 @@ const ( // AskLanguages are the build seats' languages. var AskLanguages = []string{AskGo, AskNode, AskImage} +// AskRuns is how many times an ask is run before it is taken for poison (issue 1907): an ask whose run never ends (it +// takes its node down, or its process dies with it) is delivered again by its stream to the next free node (ADR 0525 +// §6); the holder that takes it past AskRuns runs it no more, writes its step failed saying so, and ends it, so the +// asker's key fails with why and no further node is taken down. NATS-native: the count is the stream's own delivery +// count of the message, no timer. +const AskRuns = 3 + +// AskMaxDeliver is the build seats' workers' bound of deliveries (the broker's MaxDeliver): past AskRuns, so the holder +// that says the poison gets its delivery, with one more for a holder that dies saying it. +const AskMaxDeliver = AskRuns + 2 + // StepAskVersion is the version of an Ask value; a reader refuses another (ErrUnknownVersion). const StepAskVersion = 1 diff --git a/vendor/modules.txt b/vendor/modules.txt index 9a87f16c..ccbbe217 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1,4 +1,4 @@ -# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023141925-9205222a98bb +# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023145022-a9560890ee26 ## explicit; go 1.26.0 git.novox.be/novox/mesh-sdk/go git.novox.be/novox/mesh-sdk/go/announce