novox/mesh-controller
18
0
Fork 0
Code Issues Pull Requests 3 Releases Wiki Activity

Pin mesh-sdk at feat/r4g's head a9560890, re-vendor it and capture its conformance events there (R4g)

Browse Source
This commit is contained in:
jschoubben
2026-10-23 16:54:07 +02:00
parent 758c8e84e5
commit 4238318601
9 changed files with 154 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
testdata/beside/CAPTURED
Vendored
+1 -1
View File
@@ -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:
vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull.go
Vendored
+9
View File
@@ -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
vendor/git.novox.be/novox/mesh-sdk/go/buildstep/pull_fake.go
Vendored
+7 -1
View File
@@ -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 }}
}
vendor/git.novox.be/novox/mesh-sdk/go/delivery/buildrun/asks.go
Vendored
+116 -71
View File
@@ -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 `*.<ask id>` 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 `*.<ask id>` 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.
vendor/git.novox.be/novox/mesh-sdk/go/delivery/checker.go
Vendored
+6 -3
View File
@@ -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"`
vendor/git.novox.be/novox/mesh-sdk/go/delivery/stepasks.go
Vendored
+11
View File
@@ -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
vendor/modules.txt
Vendored
+1 -1
View File
@@ -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