Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e0d1de0475 | ||
|
|
db26f7e8b6 | ||
|
|
d4ae4b6790 | ||
|
|
2f6741b7f8 | ||
|
|
20a6dc4920 | ||
|
|
c4f37c279a | ||
|
|
7f89cb51de | ||
|
|
518dae5ec9 | ||
|
|
e8f8ef63fa | ||
|
|
4f4d5ca9c9 | ||
|
|
a8fbabbfc3 | ||
|
|
b9911dc1c1 | ||
|
|
e6b2df6506 |
@@ -42,11 +42,20 @@ func NodeCapacity(path, fallback, node string, logf func(string, ...any)) func()
|
|||||||
}
|
}
|
||||||
setting, _ = s[CapacitySetting].(string)
|
setting, _ = s[CapacitySetting].(string)
|
||||||
}
|
}
|
||||||
caps, err := buildrun.ParseCapacity(setting)
|
n, err := capacityOf(setting, node)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
say("%v, so %s pulls nothing", fmt.Errorf("the build capacity: %w", err), node)
|
say("%v, so %s pulls nothing", err, node)
|
||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
return (&buildrun.Places{Capacity: caps}).CapacityOf(node)
|
return n
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// capacityOf is node's capacity in a setting's `<node>=<asks>` pairs: DefaultCapacity for a node it does not name.
|
||||||
|
func capacityOf(setting, node string) (int, error) {
|
||||||
|
caps, err := buildrun.ParseCapacity(setting)
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("the build capacity: %w", err)
|
||||||
|
}
|
||||||
|
return (&buildrun.Places{Capacity: caps}).CapacityOf(node), nil
|
||||||
|
}
|
||||||
|
|||||||
+104
-9
@@ -9,6 +9,7 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -64,10 +65,17 @@ type Puller struct {
|
|||||||
PullWait time.Duration
|
PullWait time.Duration
|
||||||
Heartbeat time.Duration
|
Heartbeat time.Duration
|
||||||
Logf func(format string, args ...any)
|
Logf func(format string, args ...any)
|
||||||
// Load reads one key of the module's steps state (nil and no error when it is absent). When set, Pull reads the
|
// Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull
|
||||||
// node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with Put, so a pause outlives
|
// reads the node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a
|
||||||
// a restart.
|
// pause outlives a restart.
|
||||||
Load func(key string) ([]byte, error)
|
Load func(key string) ([]byte, error)
|
||||||
|
// PausePut writes the node's pause key: the module's pauses state (delivery.PausesState), which has no age, so a
|
||||||
|
// pause never ends by itself while the steps state drops its old keys. Put when nil.
|
||||||
|
PausePut Put
|
||||||
|
// WatchConfig, when set, follows the module's config value `capacity` (CapacitySetting) as mesh-config resolves it
|
||||||
|
// for this node, and wakes the puller on each change (watchCapacity): WatchConfigServed in a module, a test's own
|
||||||
|
// double. Until it hands a value the puller judges by Capacity.
|
||||||
|
WatchConfig func(key string, handle func(stdio.ProvisionChange) error) error
|
||||||
|
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
inHand int
|
inHand int
|
||||||
@@ -75,6 +83,9 @@ type Puller struct {
|
|||||||
paused bool
|
paused bool
|
||||||
wake chan struct{}
|
wake chan struct{}
|
||||||
once sync.Once
|
once sync.Once
|
||||||
|
// watched is the capacity the config watch handed last, used in place of Capacity once hasWatched.
|
||||||
|
watched int
|
||||||
|
hasWatched bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *Puller) init() {
|
func (p *Puller) init() {
|
||||||
@@ -113,7 +124,11 @@ func (p *Puller) setPaused(paused bool, by string) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if _, err := p.Put(key, json.RawMessage(raw)); err != nil {
|
put := p.PausePut
|
||||||
|
if put == nil {
|
||||||
|
put = p.Put
|
||||||
|
}
|
||||||
|
if _, err := put(key, json.RawMessage(raw)); err != nil {
|
||||||
return fmt.Errorf("buildstep: writing the pause of %s: %w", p.Node, err)
|
return fmt.Errorf("buildstep: writing the pause of %s: %w", p.Node, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -186,13 +201,29 @@ func (p *Puller) Running() []RunningAsk {
|
|||||||
|
|
||||||
// room says whether the puller may take one more ask now.
|
// room says whether the puller may take one more ask now.
|
||||||
func (p *Puller) room() bool {
|
func (p *Puller) room() bool {
|
||||||
capacity := 0
|
ok, _, _ := p.roomNow()
|
||||||
if p.Capacity != nil {
|
return ok
|
||||||
|
}
|
||||||
|
|
||||||
|
// roomNow is room, with the capacity and the asks in hand it was judged on.
|
||||||
|
func (p *Puller) roomNow() (ok bool, capacity, inHand int) {
|
||||||
|
p.mu.Lock()
|
||||||
|
watched, has := p.watched, p.hasWatched
|
||||||
|
p.mu.Unlock()
|
||||||
|
if has {
|
||||||
|
capacity = watched
|
||||||
|
} else if p.Capacity != nil {
|
||||||
capacity = p.Capacity()
|
capacity = p.Capacity()
|
||||||
}
|
}
|
||||||
p.mu.Lock()
|
p.mu.Lock()
|
||||||
defer p.mu.Unlock()
|
defer p.mu.Unlock()
|
||||||
return !p.paused && p.inHand < capacity
|
return !p.paused && p.inHand < capacity, capacity, p.inHand
|
||||||
|
}
|
||||||
|
|
||||||
|
// CapacityNow is the capacity the puller judges by: the config watch's last value, else Capacity.
|
||||||
|
func (p *Puller) CapacityNow() int {
|
||||||
|
_, capacity, _ := p.roomNow()
|
||||||
|
return capacity
|
||||||
}
|
}
|
||||||
|
|
||||||
// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
|
// Pull takes and runs asks until ctx ends. An ask still running then is neither ended nor acknowledged: the stream
|
||||||
@@ -203,8 +234,11 @@ func (p *Puller) Pull(ctx context.Context) error {
|
|||||||
return errors.New("buildstep: a puller needs its source, its steps' put and its run")
|
return errors.New("buildstep: a puller needs its source, its steps' put and its run")
|
||||||
}
|
}
|
||||||
if err := p.restore(); err != nil {
|
if err := p.restore(); err != nil {
|
||||||
return err // pulling on a pause that cannot be read could build on a node paused on purpose
|
// An unreadable pause degrades to not paused, said once: a puller that stops silently is the worse failure, and a
|
||||||
|
// node kept from building is held by its capacity (0), not by a pause.
|
||||||
|
p.logf("the pause of %s cannot be read: %v; pulling as not paused", p.Node, err)
|
||||||
}
|
}
|
||||||
|
go p.watchCapacity() // WatchConfigServed waits for Serve, which a module starts after its pull
|
||||||
wait := p.PullWait
|
wait := p.PullWait
|
||||||
if wait <= 0 {
|
if wait <= 0 {
|
||||||
wait = DefaultPullWait
|
wait = DefaultPullWait
|
||||||
@@ -213,12 +247,17 @@ func (p *Puller) Pull(ctx context.Context) error {
|
|||||||
if err := ctx.Err(); err != nil {
|
if err := ctx.Err(); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if !p.room() {
|
if ok, capacity, inHand := p.roomNow(); !ok {
|
||||||
|
// Parked until something wakes it: a capacity change (the config watch), a resume, or a step's end. Every
|
||||||
|
// park logs once, so a puller that sits idle says why (issue 1887).
|
||||||
|
p.logf("the %s puller on %s parks: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
case <-p.wake:
|
case <-p.wake:
|
||||||
}
|
}
|
||||||
|
_, capacity, inHand = p.roomNow()
|
||||||
|
p.logf("the %s puller on %s wakes: %d of capacity %d in hand, paused %v", p.Language, p.Node, inHand, capacity, p.Paused())
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
msg, err := p.next(ctx, wait)
|
msg, err := p.next(ctx, wait)
|
||||||
@@ -348,3 +387,59 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask delivery.StepAsk, key
|
|||||||
p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
|
p.logf("the ask %s ended on %s and its acknowledgement failed: %v", ask.ID, p.Node, aerr)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// watchCapacity follows the module's config value `capacity` when WatchConfig is set (issue 1887, 2026-10-25: g14's
|
||||||
|
// puller started at capacity 0 sat parked with capacity 2 set, because nothing woke it to read its settings file
|
||||||
|
// again). The value comes as mesh-config resolved it for this node, the event of the change itself: the runtime hands
|
||||||
|
// the value now and then each change its machine's provisioner client says moved, so nothing polls, and the puller does
|
||||||
|
// not wait for the node-engine to fill its settings file again. Each value handed is kept, used in place of Capacity,
|
||||||
|
// and wakes the puller. A watch that cannot start leaves the puller on Capacity, said once.
|
||||||
|
func (p *Puller) watchCapacity() {
|
||||||
|
if p.WatchConfig == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
err := p.WatchConfig(CapacitySetting, func(c stdio.ProvisionChange) error {
|
||||||
|
n, why := p.capacityFrom(c)
|
||||||
|
p.mu.Lock()
|
||||||
|
before, had := p.watched, p.hasWatched
|
||||||
|
p.watched, p.hasWatched = n, true
|
||||||
|
p.mu.Unlock()
|
||||||
|
switch {
|
||||||
|
case why != "":
|
||||||
|
p.logf("the %s puller on %s pulls nothing: %s", p.Language, p.Node, why)
|
||||||
|
case !had || before != n:
|
||||||
|
p.logf("the %s build capacity of %s is %d, as mesh-config resolved it", p.Language, p.Node, n)
|
||||||
|
}
|
||||||
|
p.Wake()
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
p.logf("the %s build capacity of %s cannot be followed on the bus, so it is read from the settings file when the "+
|
||||||
|
"puller wakes: %v", p.Language, p.Node, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// WatchConfigServed is stdio.WatchConfig once Serve has the runtime's channel (stdio.Served): a build module starts its
|
||||||
|
// pull before it serves, and a watch asked before that would be refused.
|
||||||
|
func WatchConfigServed(key string, handle func(stdio.ProvisionChange) error) error {
|
||||||
|
<-stdio.Served()
|
||||||
|
return stdio.WatchConfig(key, handle)
|
||||||
|
}
|
||||||
|
|
||||||
|
// capacityFrom is the node's capacity in one value the watch handed, and why it is 0 when it cannot be read.
|
||||||
|
func (p *Puller) capacityFrom(c stdio.ProvisionChange) (int, string) {
|
||||||
|
if c.Refused != "" {
|
||||||
|
return 0, "its capacity is no longer answered: " + c.Refused
|
||||||
|
}
|
||||||
|
setting := ""
|
||||||
|
if len(c.Value) > 0 && string(c.Value) != "null" {
|
||||||
|
if err := json.Unmarshal(c.Value, &setting); err != nil {
|
||||||
|
return 0, fmt.Sprintf("its capacity %s is not a string", c.Value)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
n, err := capacityOf(setting, p.Node)
|
||||||
|
if err != nil {
|
||||||
|
return 0, err.Error()
|
||||||
|
}
|
||||||
|
return n, ""
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,39 @@
|
|||||||
|
package buildstep
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
|
)
|
||||||
|
|
||||||
|
// configWatch is a double of stdio.WatchConfig: it hands the value now at the start, as the runtime does, and a test
|
||||||
|
// hands each change after with set.
|
||||||
|
type configWatch struct {
|
||||||
|
mu sync.Mutex
|
||||||
|
value string
|
||||||
|
handle func(stdio.ProvisionChange) error
|
||||||
|
asked string
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w *configWatch) watch(key string, handle func(stdio.ProvisionChange) error) error {
|
||||||
|
w.mu.Lock()
|
||||||
|
w.asked, w.handle = key, handle
|
||||||
|
value := w.value
|
||||||
|
w.mu.Unlock()
|
||||||
|
return handle(stdio.ProvisionChange{Kind: "config", Key: key, Value: json.RawMessage(`"` + value + `"`)})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w *configWatch) set(t *testing.T, value string) {
|
||||||
|
t.Helper()
|
||||||
|
w.mu.Lock()
|
||||||
|
handle, key := w.handle, w.asked
|
||||||
|
w.mu.Unlock()
|
||||||
|
if handle == nil {
|
||||||
|
t.Fatal("the puller watches no config value")
|
||||||
|
}
|
||||||
|
if err := handle(stdio.ProvisionChange{Kind: "config", Key: key, Value: json.RawMessage(`"` + value + `"`)}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
+101
-24
@@ -3,11 +3,15 @@ package buildstep
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -175,36 +179,85 @@ func (k *kv) load(key string) ([]byte, error) {
|
|||||||
return k.vals[key], nil
|
return k.vals[key], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1887/A2b, 1887/AC-zero: at capacity 0 the puller takes nothing; a capacity raised takes effect at the next pull.
|
// 1887/A2b, 1887/AC-zero, 1887/AC-wake: at capacity 0 the puller takes nothing; a capacity raised takes effect at the
|
||||||
|
// next pull, or, with the config watch (g14 idle at capacity 2, 2026-10-25), at once when the value moves, with no
|
||||||
|
// pause, resume or step end to wake it; a refused value pulls nothing, and a watch that cannot start leaves the
|
||||||
|
// settings file's capacity.
|
||||||
func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
|
func TestPullAtCapacityZeroTakesNothing(t *testing.T) {
|
||||||
src, st := NewFakeAsks(), &kv{}
|
t.Run("settings file", func(t *testing.T) {
|
||||||
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
src, st := NewFakeAsks(), &kv{}
|
||||||
var mu sync.Mutex
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||||
capacity := 0
|
var mu sync.Mutex
|
||||||
p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
|
capacity := 0
|
||||||
Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
|
||||||
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
||||||
defer cancel()
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
go p.Pull(ctx)
|
defer cancel()
|
||||||
time.Sleep(50 * time.Millisecond)
|
go p.Pull(ctx)
|
||||||
if src.Waiting(delivery.AskDelivery) != 1 {
|
time.Sleep(50 * time.Millisecond)
|
||||||
t.Fatal("a node at capacity 0 took an ask")
|
if src.Waiting(delivery.AskDelivery) != 1 {
|
||||||
|
t.Fatal("a node at capacity 0 took an ask")
|
||||||
|
}
|
||||||
|
mu.Lock()
|
||||||
|
capacity = 1
|
||||||
|
mu.Unlock()
|
||||||
|
p.Wake()
|
||||||
|
waitFor(t, "the ask taken at capacity 1", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
||||||
|
})
|
||||||
|
t.Run("config watch wakes", func(t *testing.T) {
|
||||||
|
src, st := NewFakeAsks(), newSteps()
|
||||||
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||||
|
w := &configWatch{value: "shanks=3,g14=0"}
|
||||||
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "g14", Put: st.put, PullWait: 10 * time.Millisecond,
|
||||||
|
Capacity: func() int { return 0 }, // the settings file, not refilled yet
|
||||||
|
WatchConfig: w.watch,
|
||||||
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
go p.Pull(ctx)
|
||||||
|
waitFor(t, "the watch of the capacity", func() bool { w.mu.Lock(); defer w.mu.Unlock(); return w.handle != nil })
|
||||||
|
time.Sleep(30 * time.Millisecond)
|
||||||
|
if src.Waiting(delivery.AskDelivery) != 1 {
|
||||||
|
t.Fatal("a node at capacity 0 took an ask")
|
||||||
|
}
|
||||||
|
if w.asked != CapacitySetting {
|
||||||
|
t.Fatalf("the puller watches %q, want %q", w.asked, CapacitySetting)
|
||||||
|
}
|
||||||
|
w.set(t, "shanks=3,g14=2")
|
||||||
|
waitFor(t, "the ask taken once the capacity moved to 2", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
||||||
|
})
|
||||||
|
for _, c := range []struct {
|
||||||
|
name string
|
||||||
|
watch func(string, func(stdio.ProvisionChange) error) error
|
||||||
|
room bool
|
||||||
|
}{
|
||||||
|
{"refused value leaves no room", func(key string, handle func(stdio.ProvisionChange) error) error {
|
||||||
|
return handle(stdio.ProvisionChange{Kind: "config", Key: key, Refused: "the module is no longer on g14"})
|
||||||
|
}, false},
|
||||||
|
{"watch that cannot start keeps the settings file", func(string, func(stdio.ProvisionChange) error) error {
|
||||||
|
return errors.New("no runtime")
|
||||||
|
}, true},
|
||||||
|
} {
|
||||||
|
t.Run(c.name, func(t *testing.T) {
|
||||||
|
p := &Puller{Source: NewFakeAsks(), Language: delivery.AskGo, Node: "g14", Put: newSteps().put,
|
||||||
|
Capacity: func() int { return 1 }, WatchConfig: c.watch,
|
||||||
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }}
|
||||||
|
p.watchCapacity()
|
||||||
|
if p.room() != c.room {
|
||||||
|
t.Fatalf("room = %v, want %v", p.room(), c.room)
|
||||||
|
}
|
||||||
|
})
|
||||||
}
|
}
|
||||||
mu.Lock()
|
|
||||||
capacity = 1
|
|
||||||
mu.Unlock()
|
|
||||||
p.Wake()
|
|
||||||
waitFor(t, "the ask taken at capacity 1", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1887/A2b: a paused puller takes nothing, its pause is kept on the pause key, and a puller started again on that state
|
// 1887/A2b: a paused puller takes nothing, its pause is kept on the pause key, and a puller started again on that state
|
||||||
// stays paused until resumed.
|
// stays paused until resumed.
|
||||||
func TestPullPauseOutlivesARestart(t *testing.T) {
|
func TestPullPauseOutlivesARestart(t *testing.T) {
|
||||||
src, st := NewFakeAsks(), &kv{}
|
src, st, steps := NewFakeAsks(), &kv{}, &kv{}
|
||||||
run := func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }
|
run := func(ctx context.Context, a delivery.StepAsk) (any, error) { return nil, nil }
|
||||||
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond,
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
|
||||||
Capacity: func() int { return 2 }, Run: run}
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
|
||||||
if err := p.Pause("the operator"); err != nil {
|
if err := p.Pause("the operator"); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -213,9 +266,12 @@ func TestPullPauseOutlivesARestart(t *testing.T) {
|
|||||||
if v, err := delivery.DecodePause(raw); err != nil || !v.Paused || v.By != "the operator" {
|
if v, err := delivery.DecodePause(raw); err != nil || !v.Paused || v.By != "the operator" {
|
||||||
t.Fatalf("the pause key holds %s (%v)", raw, err)
|
t.Fatalf("the pause key holds %s (%v)", raw, err)
|
||||||
}
|
}
|
||||||
|
if raw, _ := steps.load(key); raw != nil {
|
||||||
|
t.Fatalf("the pause was written in the steps state, whose keys age out: %s", raw)
|
||||||
|
}
|
||||||
// The restart: a new puller on the same state.
|
// The restart: a new puller on the same state.
|
||||||
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond,
|
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
|
||||||
Capacity: func() int { return 2 }, Run: run}
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
|
||||||
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -228,6 +284,27 @@ func TestPullPauseOutlivesARestart(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
waitFor(t, "the ask taken after the resume", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
waitFor(t, "the ask taken after the resume", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
||||||
|
|
||||||
|
t.Run("a pause that cannot be read is said and pulled as not paused", func(t *testing.T) {
|
||||||
|
src := NewFakeAsks()
|
||||||
|
var said atomic.Value
|
||||||
|
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put,
|
||||||
|
Load: func(string) ([]byte, error) { return nil, errors.New("no runtime yet") },
|
||||||
|
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run,
|
||||||
|
Logf: func(f string, a ...any) {
|
||||||
|
if m := fmt.Sprintf(f, a...); strings.Contains(m, "cannot be read") {
|
||||||
|
said.Store(m)
|
||||||
|
}
|
||||||
|
}}
|
||||||
|
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
go p.Pull(ctx)
|
||||||
|
waitFor(t, "the ask taken", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
|
||||||
|
if m, _ := said.Load().(string); !strings.Contains(m, "pulling as not paused") {
|
||||||
|
t.Fatalf("the unreadable pause was not said: %q", m)
|
||||||
|
}
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// 1887/A2c: a long step tells the bus it is still being worked; a holder that stops mid-step neither ends nor
|
// 1887/A2c: a long step tells the bus it is still being worked; a holder that stops mid-step neither ends nor
|
||||||
|
|||||||
@@ -14,11 +14,7 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool {
|
|||||||
Run: func(map[string]any) (any, error) { return run() }}
|
Run: func(map[string]any) (any, error) { return run() }}
|
||||||
}
|
}
|
||||||
current := func() (any, error) {
|
current := func() (any, error) {
|
||||||
capacity := 0
|
return map[string]any{"node": p.Node, "capacity": p.CapacityNow(), "running": p.Running(), "paused": p.Paused()}, nil
|
||||||
if p.Capacity != nil {
|
|
||||||
capacity = p.Capacity()
|
|
||||||
}
|
|
||||||
return map[string]any{"node": p.Node, "capacity": capacity, "running": p.Running(), "paused": p.Paused()}, nil
|
|
||||||
}
|
}
|
||||||
return []stdio.Tool{
|
return []stdio.Tool{
|
||||||
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity, "+
|
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity, "+
|
||||||
|
|||||||
+12
-2
@@ -181,15 +181,25 @@ type Want struct {
|
|||||||
Rule int `json:"rule,omitempty"`
|
Rule int `json:"rule,omitempty"`
|
||||||
Backup string `json:"backup,omitempty"`
|
Backup string `json:"backup,omitempty"`
|
||||||
Words string `json:"words,omitempty"`
|
Words string `json:"words,omitempty"`
|
||||||
|
// Put says the want was written by a put (ADR 0546), and Hold that the put holds it: the rules pass over it until it
|
||||||
|
// is released. A put's want names no newest fingerprint or rule.
|
||||||
|
Put bool `json:"put,omitempty"`
|
||||||
|
Hold bool `json:"hold,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func (w Want) check() error {
|
func (w Want) check() error {
|
||||||
if w.Removed {
|
if w.Removed {
|
||||||
if w.FP != "" || w.Composed != "" || w.Newest != "" || w.Rule != 0 || w.Backup != "" {
|
if w.FP != "" || w.Composed != "" || w.Newest != "" || w.Rule != 0 || w.Backup != "" || w.Put || w.Hold {
|
||||||
return fmt.Errorf("delivery: a removed want names no fingerprint, composition, rule or backup answer")
|
return fmt.Errorf("delivery: a removed want names no fingerprint, composition, rule, backup answer or put")
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
if w.Hold && !w.Put {
|
||||||
|
return fmt.Errorf("delivery: only a put holds a want")
|
||||||
|
}
|
||||||
|
if w.Put && (w.Newest != "" || w.Rule != 0) {
|
||||||
|
return fmt.Errorf("delivery: a put's want is held back by no rule")
|
||||||
|
}
|
||||||
if err := checkFingerprint(w.FP, "a want's fingerprint"); err != nil {
|
if err := checkFingerprint(w.FP, "a want's fingerprint"); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// 1297/B1a: a want key round-trips through its parser; a held key, a have key and a node's own key are refused,
|
// 1297/B1a: a want key round-trips through its parser; a held key, a have key and a node's own key are refused,
|
||||||
@@ -77,6 +78,7 @@ func TestAssignerValuesRoundTrip(t *testing.T) {
|
|||||||
{FP: fp2of('a'), Composed: hashOf('1')},
|
{FP: fp2of('a'), Composed: hashOf('1')},
|
||||||
{FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: RuleCanary, Words: "the canary ace does not run it healthy yet"},
|
{FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: RuleCanary, Words: "the canary ace does not run it healthy yet"},
|
||||||
{Removed: true, Words: "the module left g14"},
|
{Removed: true, Words: "the module left g14"},
|
||||||
|
{FP: fp2of('a'), Composed: hashOf('1'), Put: true, Hold: true, Words: "put by the operator: the bus's user list"},
|
||||||
}
|
}
|
||||||
for _, w := range wants {
|
for _, w := range wants {
|
||||||
data, err := EncodeWant(w)
|
data, err := EncodeWant(w)
|
||||||
@@ -89,6 +91,24 @@ func TestAssignerValuesRoundTrip(t *testing.T) {
|
|||||||
t.Errorf("want %+v came back %+v, %v", w, got, err)
|
t.Errorf("want %+v came back %+v, %v", w, got, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
at := time.Date(2026, 10, 25, 21, 0, 0, 0, time.UTC)
|
||||||
|
p := Put{FP: fp2of('a'), Hold: true, Why: "the bus's user list", By: "shanks.node-tools", At: at}.Said(at, PutAsked, "put asked")
|
||||||
|
p = p.Said(at, PutAsked, "put asked").Said(at.Add(time.Second), PutWanted, "want written")
|
||||||
|
p.Composed = hashOf('1')
|
||||||
|
if len(p.Steps) != 2 || p.Step != PutWanted || PutEnded(p.Step) || !PutEnded(PutReleased) {
|
||||||
|
t.Errorf("a put's steps are %+v at %s", p.Steps, p.Step)
|
||||||
|
}
|
||||||
|
pkey, _ := PutKey("novox", "nats")
|
||||||
|
if n, m, err := ParsePutKey(pkey); err != nil || n != "novox" || m != "nats" {
|
||||||
|
t.Errorf("the put key %s reads %s %s, %v", pkey, n, m, err)
|
||||||
|
}
|
||||||
|
pdata, err := EncodePut(p)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if gotP, err := DecodePut(pdata); err != nil || gotP.FP != p.FP || gotP.Step != PutWanted || len(gotP.Steps) != 2 || !gotP.Hold {
|
||||||
|
t.Errorf("the put came back %+v, %v", gotP, err)
|
||||||
|
}
|
||||||
h := Held{Node: "ace", Words: "its health check failed: connection refused"}
|
h := Held{Node: "ace", Words: "its health check failed: connection refused"}
|
||||||
data, err := EncodeHeld(h)
|
data, err := EncodeHeld(h)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -128,6 +148,9 @@ func TestAssignerValuesRefused(t *testing.T) {
|
|||||||
{FP: fp2of('a'), Newest: fp2of('b')},
|
{FP: fp2of('a'), Newest: fp2of('b')},
|
||||||
{FP: fp2of('a'), Newest: fp2of('a'), Rule: 3},
|
{FP: fp2of('a'), Newest: fp2of('a'), Rule: 3},
|
||||||
{FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: 8},
|
{FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: 8},
|
||||||
|
{FP: fp2of('a'), Composed: hashOf('1'), Hold: true},
|
||||||
|
{FP: fp2of('a'), Composed: hashOf('1'), Put: true, Newest: fp2of('b'), Rule: 3},
|
||||||
|
{Removed: true, Put: true},
|
||||||
} {
|
} {
|
||||||
if _, err := EncodeWant(w); err == nil {
|
if _, err := EncodeWant(w); err == nil {
|
||||||
t.Errorf("EncodeWant(%+v) accepted", w)
|
t.Errorf("EncodeWant(%+v) accepted", w)
|
||||||
@@ -148,6 +171,19 @@ func TestAssignerValuesRefused(t *testing.T) {
|
|||||||
t.Errorf("EncodeDifference(%+v) accepted", d)
|
t.Errorf("EncodeDifference(%+v) accepted", d)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
for _, p := range []Put{
|
||||||
|
{FP: fp2of('a'), Why: "w", Step: "walking"},
|
||||||
|
{FP: fp2of('a'), Step: PutAsked},
|
||||||
|
{FP: "src2:abc", Why: "w", Step: PutAsked},
|
||||||
|
{FP: fp2of('a'), Why: "w", Step: PutWanted, Composed: "sha256:x"},
|
||||||
|
} {
|
||||||
|
if _, err := EncodePut(p); err == nil {
|
||||||
|
t.Errorf("EncodePut(%+v) accepted", p)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if _, _, err := ParsePutKey("ace.node"); err == nil {
|
||||||
|
t.Errorf("a node's own key was read as a put")
|
||||||
|
}
|
||||||
if _, err := DecodeWant([]byte(`{"v":2,"fp":"` + fp2of('a') + `"}`)); !errors.Is(err, ErrUnknownVersion) {
|
if _, err := DecodeWant([]byte(`{"v":2,"fp":"` + fp2of('a') + `"}`)); !errors.Is(err, ErrUnknownVersion) {
|
||||||
t.Errorf("a want of version 2 was read: %v", err)
|
t.Errorf("a want of version 2 was read: %v", err)
|
||||||
}
|
}
|
||||||
@@ -262,4 +298,20 @@ func TestControllerGrant(t *testing.T) {
|
|||||||
if _, err := EncodeControllerGrant(ControllerGrant{Accounts: []string{""}}); err == nil {
|
if _, err := EncodeControllerGrant(ControllerGrant{Accounts: []string{""}}); err == nil {
|
||||||
t.Errorf("a grant naming an empty account was accepted")
|
t.Errorf("a grant naming an empty account was accepted")
|
||||||
}
|
}
|
||||||
|
g := ControllerGrant{Accounts: []string{"ace.mesh-build-go", "shanks.mesh-mail.reader"}}
|
||||||
|
for _, c := range []struct {
|
||||||
|
g ControllerGrant
|
||||||
|
node, module string
|
||||||
|
names bool
|
||||||
|
}{
|
||||||
|
{g, "ace", "mesh-build-go", true},
|
||||||
|
{g, "shanks", "mesh-mail", true},
|
||||||
|
{g, "g14", "mesh-build-go", false},
|
||||||
|
{g, "ace", "mesh-build", false},
|
||||||
|
{ControllerGrant{}, "novox", "nats", true},
|
||||||
|
} {
|
||||||
|
if got := c.g.NamesNode(c.node, c.module); got != c.names {
|
||||||
|
t.Errorf("%v names %s.%s: %v, want %v", c.g.Accounts, c.node, c.module, got, c.names)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -73,3 +73,20 @@ func DecodeControllerGrant(data []byte) (ControllerGrant, error) {
|
|||||||
|
|
||||||
// SameControllerGrant says whether two grants are the same value.
|
// SameControllerGrant says whether two grants are the same value.
|
||||||
func SameControllerGrant(a, b ControllerGrant) bool { return reflect.DeepEqual(a, b) }
|
func SameControllerGrant(a, b ControllerGrant) bool { return reflect.DeepEqual(a, b) }
|
||||||
|
|
||||||
|
// NamesNode says whether the grant names module's accounts on node: its own `<node>.<module>` or a process's
|
||||||
|
// `<node>.<module>.<process>`. A grant that names no account (a module that does not speak on the bus) names every
|
||||||
|
// node. The assigner wants a node only once the grant names its account (issue 1945), and a put waits for it (ADR
|
||||||
|
// 0546 clause 4).
|
||||||
|
func (g ControllerGrant) NamesNode(node, module string) bool {
|
||||||
|
if len(g.Accounts) == 0 {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
own := node + "." + module
|
||||||
|
for _, a := range g.Accounts {
|
||||||
|
if a == own || strings.HasPrefix(a, own+".") {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|||||||
@@ -0,0 +1,148 @@
|
|||||||
|
package delivery
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// A put (novox/hq ADR 0546): one version of a module put on one node by the assigner's tool `assigner_put`. Each put is the
|
||||||
|
// key `<node>.<module>` of the assigner's state puts, which holds the version asked, the hold, why, and each step the
|
||||||
|
// assigner took or waits on, so a reader watches the key to follow it. The want the put writes carries Put (and Hold),
|
||||||
|
// so the assigner's rules pass over that key while the put holds it.
|
||||||
|
|
||||||
|
// AssignerPutsBucket is the assigner's bucket of puts, `<owner module>_puts`.
|
||||||
|
const AssignerPutsBucket = AssignerModule + "_puts"
|
||||||
|
|
||||||
|
// AssignerPutsHistory is the revisions the puts bucket keeps per key: each step of a put is one.
|
||||||
|
const AssignerPutsHistory = 20
|
||||||
|
|
||||||
|
// A put's steps, in the order they are taken (ADR 0546 clauses 3 to 9). A put stands at the last step written.
|
||||||
|
const (
|
||||||
|
// PutAsked is a put written by the tool, not taken up by a pass yet.
|
||||||
|
PutAsked = "asked"
|
||||||
|
// PutBuild waits for the builder's build at the fingerprint.
|
||||||
|
PutBuild = "build"
|
||||||
|
// PutCompose waits for the controller's composition (and grant) at the fingerprint.
|
||||||
|
PutCompose = "compose"
|
||||||
|
// PutPerson waits for the operator's answer to the ask about the move.
|
||||||
|
PutPerson = "person"
|
||||||
|
// PutSnapshot waits for the bus's streams to be snapshotted after the operator's yes.
|
||||||
|
PutSnapshot = "snapshot"
|
||||||
|
// PutWanted is the want written: the node's have has not reached it yet.
|
||||||
|
PutWanted = "wanted"
|
||||||
|
// PutHolding is a held put whose have runs it healthy: the want stays until released.
|
||||||
|
PutHolding = "holding"
|
||||||
|
// PutDone is a put without hold whose have ran it healthy: the rules write the want again.
|
||||||
|
PutDone = "done"
|
||||||
|
// PutReleased is a put released by the tool `assigner_release`.
|
||||||
|
PutReleased = "released"
|
||||||
|
// PutRefused is a put that cannot go on: a build that failed, an answer no, a snapshot not taken. Its words say why.
|
||||||
|
PutRefused = "refused"
|
||||||
|
)
|
||||||
|
|
||||||
|
// PutEnded says a put at this step is over: nothing more is done for it and the rules own the want.
|
||||||
|
func PutEnded(step string) bool {
|
||||||
|
return step == PutDone || step == PutReleased || step == PutRefused
|
||||||
|
}
|
||||||
|
|
||||||
|
// PutStep is one step a put took: when, which, and what was done or waits, in a sentence.
|
||||||
|
type PutStep struct {
|
||||||
|
At time.Time `json:"at"`
|
||||||
|
Step string `json:"step"`
|
||||||
|
Words string `json:"words"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// PutStepsKept is how many steps a put keeps, newest last.
|
||||||
|
const PutStepsKept = 16
|
||||||
|
|
||||||
|
// Put is the value of a key of AssignerPutsBucket.
|
||||||
|
type Put struct {
|
||||||
|
V int `json:"v"`
|
||||||
|
// FP is the fingerprint put, of the new form.
|
||||||
|
FP string `json:"fp"`
|
||||||
|
// Hold keeps the want at FP until released.
|
||||||
|
Hold bool `json:"hold,omitempty"`
|
||||||
|
Why string `json:"why"`
|
||||||
|
// By is who asked, as the bus named the caller; "unproven" when it named none.
|
||||||
|
By string `json:"by,omitempty"`
|
||||||
|
At time.Time `json:"at"`
|
||||||
|
// Step is where the put stands now, one of the Put* steps; Words say it in a sentence.
|
||||||
|
Step string `json:"step"`
|
||||||
|
Words string `json:"words,omitempty"`
|
||||||
|
// Composed is the digest of the composition the want names, once known.
|
||||||
|
Composed string `json:"composed,omitempty"`
|
||||||
|
// Ask is the key, in AssignerAsksBucket, of the ask about the move, once asked.
|
||||||
|
Ask string `json:"ask,omitempty"`
|
||||||
|
// Snapshot says where the bus's streams were snapshotted, once they were.
|
||||||
|
Snapshot string `json:"snapshot,omitempty"`
|
||||||
|
// Steps are the steps taken, oldest first, at most PutStepsKept.
|
||||||
|
Steps []PutStep `json:"steps,omitempty"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Said is p at step with words, the step added to its steps when it differs from the last one.
|
||||||
|
func (p Put) Said(at time.Time, step, words string) Put {
|
||||||
|
if p.Step == step && p.Words == words {
|
||||||
|
return p
|
||||||
|
}
|
||||||
|
p.Step, p.Words = step, words
|
||||||
|
p.Steps = append(append([]PutStep(nil), p.Steps...), PutStep{At: at.UTC(), Step: step, Words: words})
|
||||||
|
if len(p.Steps) > PutStepsKept {
|
||||||
|
p.Steps = p.Steps[len(p.Steps)-PutStepsKept:]
|
||||||
|
}
|
||||||
|
return p
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p Put) check() error {
|
||||||
|
if err := checkFingerprint(p.FP, "a put's fingerprint"); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if p.Why == "" {
|
||||||
|
return fmt.Errorf("delivery: a put says why")
|
||||||
|
}
|
||||||
|
switch p.Step {
|
||||||
|
case PutAsked, PutBuild, PutCompose, PutPerson, PutSnapshot, PutWanted, PutHolding, PutDone, PutReleased, PutRefused:
|
||||||
|
default:
|
||||||
|
return fmt.Errorf("delivery: %q is not a step of a put", p.Step)
|
||||||
|
}
|
||||||
|
if p.Composed != "" {
|
||||||
|
return checkDigest(p.Composed, "a put's composition digest")
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// PutKey is the key, in AssignerPutsBucket, of the put of module on node: `<node>.<module>`.
|
||||||
|
func PutKey(node, module string) (string, error) {
|
||||||
|
return HaveKey(node, module)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ParsePutKey reads a PutKey.
|
||||||
|
func ParsePutKey(key string) (node, module string, err error) {
|
||||||
|
node, module, isNode, err := ParseHaveKey(key)
|
||||||
|
if err == nil && isNode {
|
||||||
|
err = fmt.Errorf("delivery: %q is a node's own key, and no put", key)
|
||||||
|
}
|
||||||
|
return node, module, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// EncodePut is p as JSON at Version; it refuses a fingerprint not of the new form, a put that says no why and an
|
||||||
|
// unknown step.
|
||||||
|
func EncodePut(p Put) ([]byte, error) {
|
||||||
|
if err := p.check(); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
p.V = Version
|
||||||
|
return json.Marshal(p)
|
||||||
|
}
|
||||||
|
|
||||||
|
// DecodePut reads a put; it refuses an unknown version (ErrUnknownVersion) and what EncodePut refuses.
|
||||||
|
func DecodePut(data []byte) (Put, error) {
|
||||||
|
var p Put
|
||||||
|
if err := decode(data, &p); err != nil {
|
||||||
|
return Put{}, err
|
||||||
|
}
|
||||||
|
if err := p.check(); err != nil {
|
||||||
|
return Put{}, err
|
||||||
|
}
|
||||||
|
return p, nil
|
||||||
|
}
|
||||||
@@ -535,9 +535,13 @@ func ParseAskStepKey(key string) (node, askID string, err error) {
|
|||||||
return node, askID, nil
|
return node, askID, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// PauseKey is the key, in a build module's StepsBucket, that keeps whether the module on node takes asks:
|
// PausesState is the state a build module keeps its nodes' pause keys in. It has no age, unlike StepsState, whose keys
|
||||||
|
// are per step and dropped after their age: a pause kept among the steps would end by itself.
|
||||||
|
const PausesState = "pauses"
|
||||||
|
|
||||||
|
// PauseKey is the key, in a build module's PausesState, that keeps whether the module on node takes asks:
|
||||||
// `paused.<node>` (issue 1887's design limit 2: every holder pulls from one shared worker, so a node's pause is its holder
|
// `paused.<node>` (issue 1887's design limit 2: every holder pulls from one shared worker, so a node's pause is its holder
|
||||||
// not fetching, kept here so a restart keeps it). A pause key is never an ask's step key: its second part is a node.
|
// not fetching, kept so a restart keeps it). A pause key is never an ask's step key: its second part is a node.
|
||||||
func PauseKey(node string) (string, error) {
|
func PauseKey(node string) (string, error) {
|
||||||
if err := checkName("node", node); err != nil {
|
if err := checkName("node", node); err != nil {
|
||||||
return "", err
|
return "", err
|
||||||
|
|||||||
@@ -283,9 +283,15 @@ func ViewAnswerFits(a ViewAnswer, ask AssignerAsk) error {
|
|||||||
if a.Asking != ask.Ask.ID {
|
if a.Asking != ask.Ask.ID {
|
||||||
return fmt.Errorf("delivery: the view's answer is to the asking %s, and the ask stands at the asking %s", a.Asking, ask.Ask.ID)
|
return fmt.Errorf("delivery: the view's answer is to the asking %s, and the ask stands at the asking %s", a.Asking, ask.Ask.ID)
|
||||||
}
|
}
|
||||||
if _, ok := ask.Ask.Option(a.Option); !ok {
|
o, ok := ask.Ask.Option(a.Option)
|
||||||
|
if !ok {
|
||||||
return fmt.Errorf("delivery: the view's answer chose %q, which the asking %s does not offer", a.Option, ask.Ask.ID)
|
return fmt.Errorf("delivery: the view's answer chose %q, which the asking %s does not offer", a.Option, ask.Ask.ID)
|
||||||
}
|
}
|
||||||
|
if o.Level != "" && o.Level != asks.Acknowledge {
|
||||||
|
// An option that authorises is answered on a channel that proves who answered (ADR 0259 §5, ADR 0546 clause 8),
|
||||||
|
// never in the view, which proves nobody.
|
||||||
|
return fmt.Errorf("delivery: the view's answer chose %q, an option of level %s, which only a channel that proves who answered takes", a.Option, o.Level)
|
||||||
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -158,4 +158,11 @@ func TestViewAnswerFits(t *testing.T) {
|
|||||||
if err := ViewAnswerFits(ViewAnswer{Asking: first.Ask.ID, Option: "restore"}, first); err == nil {
|
if err := ViewAnswerFits(ViewAnswer{Asking: first.Ask.ID, Option: "restore"}, first); err == nil {
|
||||||
t.Errorf("an option the ask does not offer fits")
|
t.Errorf("an option the ask does not offer fits")
|
||||||
}
|
}
|
||||||
|
// An option that authorises is never answered in the view (ADR 0546 clause 8).
|
||||||
|
approving := first
|
||||||
|
approving.Ask.Options = append([]asks.Option(nil), first.Ask.Options...)
|
||||||
|
approving.Ask.Options[0].Level = asks.Approve
|
||||||
|
if err := ViewAnswerFits(answer, approving); err == nil || !strings.Contains(err.Error(), "approve") {
|
||||||
|
t.Errorf("the view answered an option of level approve: %v", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,11 +1,16 @@
|
|||||||
package process
|
package process
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/nats-io/nats.go/jetstream"
|
||||||
)
|
)
|
||||||
|
|
||||||
// 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node,
|
// 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node,
|
||||||
@@ -46,6 +51,28 @@ func TestReadAccountToleratesNoClaimsAtAll(t *testing.T) {
|
|||||||
// 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears
|
// 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears
|
||||||
// in the error even when it is the field present and another is missing.
|
// in the error even when it is the field present and another is missing.
|
||||||
func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
|
func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
|
||||||
|
t.Run("1945: a refused bucket stops a start only when the part cannot work without it", func(t *testing.T) {
|
||||||
|
refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_mesh-delivery-gate_source")
|
||||||
|
for _, c := range []struct {
|
||||||
|
name string
|
||||||
|
err error
|
||||||
|
}{{"refused", refusal}, {"not answered", context.DeadlineExceeded}} {
|
||||||
|
var said []string
|
||||||
|
logf := func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
|
||||||
|
if kv := OptionalBucket(context.Background(), refusingBus{c.err}, "b", "b is not heard", logf); kv != nil {
|
||||||
|
t.Errorf("%s: an optional bucket answered %v", c.name, kv)
|
||||||
|
}
|
||||||
|
NotWatched(logf, "b", c.err, "b is not heard")
|
||||||
|
if len(said) != 2 || !strings.Contains(said[0], c.err.Error()) || !strings.Contains(said[1], "cannot be watched") {
|
||||||
|
t.Errorf("%s: said %q", c.name, said)
|
||||||
|
}
|
||||||
|
if _, err := EssentialBucket(context.Background(), refusingBus{c.err}, "builder", "b", "build"); err == nil ||
|
||||||
|
!strings.Contains(err.Error(), "grant of b") || !strings.Contains(err.Error(), "the builder cannot build") {
|
||||||
|
t.Errorf("%s: an essential bucket: %v", c.name, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
for _, tc := range []struct {
|
for _, tc := range []struct {
|
||||||
field string
|
field string
|
||||||
values map[string]any
|
values map[string]any
|
||||||
@@ -82,3 +109,8 @@ func writeAccount(t *testing.T, fields map[string]any) string {
|
|||||||
}
|
}
|
||||||
return path
|
return path
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// refusingBus refuses every bucket with err.
|
||||||
|
type refusingBus struct{ err error }
|
||||||
|
|
||||||
|
func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
|
||||||
|
|||||||
@@ -0,0 +1,55 @@
|
|||||||
|
package process
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
|
||||||
|
"github.com/nats-io/nats.go/jetstream"
|
||||||
|
)
|
||||||
|
|
||||||
|
// **A part does not stop for a bucket it can work without** (novox/hq issue 1945, met twice on 2026-10-25: the delivery
|
||||||
|
// builder exited on a refused read of the gate's source and crash-looped). A grant reaches the bus with a delivery, so a
|
||||||
|
// bucket, a watch or a consumer the bus refuses, or times out on, at start is often only not granted yet: exiting for it
|
||||||
|
// stops the part until a person's hand. A part opens what it cannot do its job at all without with EssentialBucket, which
|
||||||
|
// stops its start with an error naming the bucket and its grant, and everything else with OptionalBucket, which says what
|
||||||
|
// is not heard and goes on; a watch the bus refuses on an optional bucket is said with NotWatched. Here, beside State,
|
||||||
|
// because this is where a process opens what it keeps and reads on the bus; the parts that open buckets with their own
|
||||||
|
// JetStream (the delivery parts) take the same three.
|
||||||
|
|
||||||
|
// BucketOpener is the part of the bus a start opens buckets with: a jetstream.JetStream, or a test's fake.
|
||||||
|
type BucketOpener interface {
|
||||||
|
KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
// EssentialBucket opens a bucket part cannot work without (it cannot `does` without it); its error is EssentialErr's.
|
||||||
|
func EssentialBucket(ctx context.Context, js BucketOpener, part, name, does string) (jetstream.KeyValue, error) {
|
||||||
|
kv, err := js.KeyValue(ctx, name)
|
||||||
|
if err != nil {
|
||||||
|
return nil, EssentialErr(part, name, does, err)
|
||||||
|
}
|
||||||
|
return kv, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// EssentialErr is the error of a bucket, consumer or stream named name that part cannot work without: it says what part
|
||||||
|
// cannot do, that it stops, and the grant to look at.
|
||||||
|
func EssentialErr(part, name, does string, err error) error {
|
||||||
|
return fmt.Errorf("%s cannot be read (%v): the %s cannot %s without it, so it stops; where the bus refused it or did not "+
|
||||||
|
"answer, this module's grant of %s (its manifest's reads, state or consumes, issued with its credential) has not "+
|
||||||
|
"reached the bus yet", name, err, part, does, name)
|
||||||
|
}
|
||||||
|
|
||||||
|
// OptionalBucket opens a bucket the part works without: any error is said with what is not heard (without), and nil
|
||||||
|
// answered.
|
||||||
|
func OptionalBucket(ctx context.Context, js BucketOpener, name, without string, logf func(string, ...any)) jetstream.KeyValue {
|
||||||
|
kv, err := js.KeyValue(ctx, name)
|
||||||
|
if err != nil {
|
||||||
|
logf("the bucket %s cannot be read (%v): %s", name, err, without)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return kv
|
||||||
|
}
|
||||||
|
|
||||||
|
// NotWatched says that an optional bucket opened but its watch was refused or not answered: what is not heard.
|
||||||
|
func NotWatched(logf func(string, ...any), name string, err error, without string) {
|
||||||
|
logf("the bucket %s cannot be watched (%v): %s", name, err, without)
|
||||||
|
}
|
||||||
@@ -154,8 +154,15 @@ const CodeChanged = -32010
|
|||||||
var (
|
var (
|
||||||
current *channel
|
current *channel
|
||||||
currentMu sync.Mutex
|
currentMu sync.Mutex
|
||||||
|
// served is closed once Serve has the runtime's channel (Served).
|
||||||
|
served = make(chan struct{})
|
||||||
|
servedOnce sync.Once
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Served is closed once Serve has the runtime's channel, so code started before Serve can wait for it instead of
|
||||||
|
// asking again until the runtime takes it: a watch of a config value, a subscribe.
|
||||||
|
func Served() <-chan struct{} { return served }
|
||||||
|
|
||||||
// Serve answers the runtime on stdin and stdout until stdin closes. An empty name serves as the
|
// Serve answers the runtime on stdin and stdout until stdin closes. An empty name serves as the
|
||||||
// module the runtime named in MESH_SERVED_MODULE.
|
// module the runtime named in MESH_SERVED_MODULE.
|
||||||
func Serve(name string, tools []Tool) error {
|
func Serve(name string, tools []Tool) error {
|
||||||
@@ -180,6 +187,7 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error {
|
|||||||
currentMu.Lock()
|
currentMu.Lock()
|
||||||
current = c
|
current = c
|
||||||
currentMu.Unlock()
|
currentMu.Unlock()
|
||||||
|
servedOnce.Do(func() { close(served) })
|
||||||
// What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so
|
// What the bundle said it keeps is announced once it serves: the runtime answers the publish on this channel, so
|
||||||
// it is asked from beside the loop that reads the answer.
|
// it is asked from beside the loop that reads the answer.
|
||||||
go func() { _ = announce1() }()
|
go func() { _ = announce1() }()
|
||||||
|
|||||||
@@ -16,6 +16,22 @@ func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
|
|||||||
t.Setenv(ServedAs, "chime")
|
t.Setenv(ServedAs, "chime")
|
||||||
t.Setenv("MESH_MODULE", "chime")
|
t.Setenv("MESH_MODULE", "chime")
|
||||||
t.Setenv("MESH_NODE", "anchor")
|
t.Setenv("MESH_NODE", "anchor")
|
||||||
|
// 1887/AC-wake: Served closes once Serve has the runtime's channel, not before; the globals are put back after.
|
||||||
|
oldServed := served
|
||||||
|
served, servedOnce = make(chan struct{}), sync.Once{}
|
||||||
|
t.Cleanup(func() {
|
||||||
|
served, servedOnce = oldServed, sync.Once{}
|
||||||
|
select {
|
||||||
|
case <-oldServed: // it was closed: its Once is spent too
|
||||||
|
servedOnce.Do(func() {})
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
})
|
||||||
|
select {
|
||||||
|
case <-Served():
|
||||||
|
t.Fatal("Served closed before Serve")
|
||||||
|
default:
|
||||||
|
}
|
||||||
inR, inW := io.Pipe()
|
inR, inW := io.Pipe()
|
||||||
outR, outW := io.Pipe()
|
outR, outW := io.Pipe()
|
||||||
go func() {
|
go func() {
|
||||||
@@ -50,6 +66,11 @@ func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
|
|||||||
if got := next()["result"].(map[string]any)["serverInfo"].(map[string]any)["name"]; got != "chime" {
|
if got := next()["result"].(map[string]any)["serverInfo"].(map[string]any)["name"]; got != "chime" {
|
||||||
t.Fatalf("served as %v, want the module the runtime named", got)
|
t.Fatalf("served as %v, want the module the runtime named", got)
|
||||||
}
|
}
|
||||||
|
select {
|
||||||
|
case <-Served():
|
||||||
|
case <-time.After(2 * time.Second):
|
||||||
|
t.Fatal("Served did not close once serving")
|
||||||
|
}
|
||||||
send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "ring", "arguments": map[string]any{}}})
|
send(map[string]any{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": map[string]any{"name": "ring", "arguments": map[string]any{}}})
|
||||||
asked := next()
|
asked := next()
|
||||||
if asked["method"] != "mesh/publish" {
|
if asked["method"] != "mesh/publish" {
|
||||||
|
|||||||
Reference in New Issue
Block a user