Compare commits

...
Author SHA1 Message Date
jschoubben e0d1de0475 process: a part opens what it cannot work without as essential, the rest as optional (issue 1945)
mesh/delivery merged: source.main moved to novox/mesh-sdk@7ac4493ce9ab
mesh/delivery-group group fix/1945-ask-again-and-why merged: every member is merged
mesh/check pass e0d1de047585 on 3db051be57ca | unit 1642 run, 752 kept, 2 skipped | integration 1710 run, 803 kept, 2 skipped | over bound: unit, inte…
The delivery builder exited twice on 2026-10-25 for a bucket the bus
refused at start, though it could build without it: a grant reaches the
bus with a delivery, so a refusal at start is often only a grant not
there yet. EssentialBucket stops a start naming the bucket and its grant;
OptionalBucket and NotWatched say what is not heard and go on. Here so
the seven delivery parts take one copy, not seven.
2026-10-25 22:12:20 +01:00
mesh-tools db26f7e8b6 Merge pull request 'The put's words: delivery.Put, a want's put and hold marks, a grant that says which node's accounts it names (ADR 0546)' (#82) from feat/put-version into main 2026-10-25 20:35:28 +00:00
jschoubben d4ae4b6790 Say the put's tools, not a seat's verbs, in the put's words (ADR 0546)
mesh/delivery merged: source.main moved to novox/mesh-sdk@db26f7e8b6f5
mesh/check pass d4ae4b6790b8 on db26f7e8b6f5 | unit 1102 run, 680 kept, 2 skipped | integration 1103 run, 722 kept, 2 skipped | over bound: unit (slow…
2026-10-25 21:08:10 +01:00
jschoubben 2f6741b7f8 No seat and no compose verb for the put: the controller takes no change, so the assigner serves the put as its own tools from compositions the controller already writes (ADR 0546) 2026-10-25 21:07:58 +01:00
jschoubben 20a6dc4920 A grant says whether it names a module's accounts on a node, for the assigner and the compose verb (ADR 0546) 2026-10-25 20:50:51 +01:00
jschoubben c4f37c279a The assigner's seat with put, release and puts, the controller's compose verb, a put's key and value, and the view answers no option that authorises (ADR 0546) 2026-10-25 20:49:29 +01:00
jschoubben 7f89cb51de Merge pull request 'Keep a build node's pause in a state of its own, and pull as not paused when the pause cannot be read (steps bucket full, 2026-10-25)' (#81) from fix/steps-bucket-ages into main
Reviewed-on: #81
2026-10-25 18:06:32 +00:00
jschoubben 518dae5ec9 Pull as not paused when the pause cannot be read, and say so once, instead of ending the puller silently
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/steps-bucket-ages held: a member is held: novox/mesh-sdk@518dae5ec9c8 (merged without a passing check: only a person decides what…
mesh/check checking 518dae5ec9c8 on 7f89cb51de15 | novox/mesh-sdk#81 in group fix/steps-bucket-ages
A Load error ended Pull, and the modules run it in a goroutine whose error nobody reads: every
build node would stop pulling when the new pauses state could not be read (or when Pull reached
restore before Serve). A node kept from building is held by capacity 0, not by a pause.
2026-10-25 18:56:17 +01:00
jschoubben e8f8ef63fa Keep a build node's pause in a pauses state of its own, so the steps state can drop its old keys without ending a pause
The steps state of the build modules filled to the bus's 64 MiB cap (27,370 keys of mesh-build-go,
none ever removed) and refused a passed check's end. Its keys get an age in the catalogue; the
pause key lived among them and would then have ended by itself. Puller.PausePut writes it to
delivery.PausesState instead; Put stays the fallback for a module not yet moved.
2026-10-25 18:51:04 +01:00
jschoubben 4f4d5ca9c9 Merge pull request 'Wake a parked build puller when its capacity changes, so a node raised from capacity 0 takes asks (issue 1887)' (#80) from fix/puller-wakes-on-capacity into main
Reviewed-on: #80
2026-10-25 17:10:33 +00:00
jschoubben a8fbabbfc3 Fold the puller-wake and Served tests into existing tests to stay within the test budget (issue 1887)
mesh/delivery merged: source.main moved to novox/mesh-sdk@4f4d5ca9c956
mesh/delivery-group group fix/puller-wakes-on-capacity merged: every member is merged
mesh/check pass a8fbabbfc373 on 4f4d5ca9c956 | unit 1102 run, 679 kept, 2 skipped | integration 1103 run, 721 kept, 2 skipped | over bound: unit, inte…
2026-10-25 17:44:03 +01:00
jschoubben b9911dc1c1 Start the puller's capacity watch once Serve has the runtime's channel, since a build module pulls before it serves (issue 1887)
A watch asked before Serve is refused, which would have left every build
module on its settings file again. stdio.Served closes when Serve has the
channel; WatchConfigServed waits on it, and Pull starts the watch beside its
loop.
2026-10-25 17:26:10 +01:00
jschoubben e6b2df6506 Wake a parked build puller on its capacity's own change, so a node started at capacity 0 takes asks once raised (issue 1887)
g14's Go builder sat idle at capacity 2 while delivery builds waited: a puller
with no room waited on its wake alone, which only a pause, a resume or a step's
end gives, and it read its capacity from the settings file only when woken.
The puller now follows the module's config value capacity through the
runtime's watch of what mesh-config resolved (stdio.WatchConfig), uses the
value it hands, and wakes on each change: no timer, no file watch, and no wait
for the node-engine to fill the file again. It says when it parks and when it
wakes, with its capacity.
2026-10-25 17:23:28 +01:00
jschoubben f9d11acf23 Merge pull request 'A node's own composition may carry filled paths; a composition that would leave a provisioner file empty is refused (issue 1944)' (#79) from fix/r6-node-composition into main
Reviewed-on: #79
2026-10-25 00:17:36 +00:00
jschoubben 10b6e20232 Let the node's own composition carry its files' filled paths, and refuse a composition that would have a provisioner's file written empty (issue 1945)
mesh/delivery held for a person: merged without a passing check: only a person decides what follows
mesh/delivery-group group fix/r6-node-composition held: a member is held: novox/mesh-sdk@10b6e202320f (merged without a passing check: only a person decides wh…
mesh/check checking 10b6e202320f on da7f0455759a
2026-10-25 01:46:36 +02:00
jschoubben da7f045575 Merge pull request 'R6: the composition's layout, its signature and digest, the want's digest, the judgement values; the hand tool is mesh-hand-compose (issue 1943)' (#78) from feat/r6 into main
Reviewed-on: #78
2026-10-24 23:17:32 +00:00
jschoubben 748c564cdf Merge main (the R4g finish: the composer command) into feat/r6 (issue 1943)
mesh/delivery delivering: 0 machine step(s) passed
mesh/delivery-group group feat/r6 merged: every member is merged
mesh/check failed: the composition: the composer of the mesh-sdk mesh-controller at bdb8193456e17c9df59215c7f3e69a83a2e5a262 pins did not run: failed …
main's mesh-compose is the composer command (ADR 0482 clause 4) and keeps its name, which composer.Package and the
build steps name; feat/r6's hand path tool, which composes one module for one node and signs it with the operator's key
(ADR 0539 clause 10), moves to cmd/mesh-hand-compose.
2026-10-24 23:01:18 +02:00
jschoubben d7a9b9688b Let the judgement beside a composition say unchanged: equal to the declaration the node last applied (issue 1943; ADR 0540 clause 9)
The first want at the switch skips the record hold only where the composition changes nothing on the node.
2026-10-24 21:17:14 +02:00
jschoubben 7f9cc0e55c Keep plan-moved and sent among the controller seat's events until the catalogue's main consumes them no more (issue 1943)
The controller's broker tests read the catalogue's main, whose gate and messenger still consume plan-moved; nothing
emits them (ADR 0539).
2026-10-24 20:38:24 +02:00
jschoubben e45bdc0891 Drop plan-moved and sent from the controller seat's events (issue 1943; the review's notes on 1945)
No plan is kept or sent in release 6 (ADR 0539), and the messenger now hears the delivery view's events instead.
2026-10-24 20:20:48 +02:00
jschoubben a523a8cbb0 Say beside a composition that it changes only the bus's user list (issue 1943; the review's A2, for issue 1906)
A revision that changes only the user list is written into the bus node's want at once, outside the record hold
(ADR 0539 clause 7), so the assigner must tell it apart from a move that recreates or moves data.
2026-10-24 18:15:32 +02:00
jschoubben 5da87a0dda Let the controller say a bucket it deleted at its first start (issue 1943; for issue 1945, B13)
ADR 0539 clause 13 has the controller delete the plan-state buckets once and say each as its event.
2026-10-24 17:03:54 +02:00
jschoubben e64486b0e6 Sign a composition through a signer that keeps its private half (issue 1943, S5)
The controller's identity store signs a message and never hands its key out, as it signed declarations.
2026-10-24 16:42:41 +02:00
jschoubben 64a502d16a Carry the module's upgrade policy and its source beside a composition's judgement (issue 1943; for issue 1906)
The assigner holds a move of a module of policy record, and only the controller knows a person's own choice of
policy (ADR 0539 clause 14).
2026-10-24 16:40:45 +02:00
jschoubben de5b33843f Run the captured declarations' test per node in parallel, inside the unit level's bound (issue 1943) 2026-10-24 16:36:42 +02:00
jschoubben d8da8f1d36 Sign a composition with a trusted key and compose one by hand with mesh-compose (issue 1943, S5)
The node-engine applies only a composition signed by the controller or the operator, and when the controller cannot
compose its own fix the operator composes, signs and writes one at the terminal (ADR 0539 clauses 3 and 10).
2026-10-24 16:36:14 +02:00
jschoubben d506f2196a Refuse the plan verbs by name on the controller and delivery seats (issue 1943, S4)
With no delivery plan nothing starts, stops, retries or reads one; each removed verb stays a row with no input so a
caller is told its replacement, optional on the delivery seat so the gate of before and of release 6 both hold it
(ADR 0539 clauses 15, 17 and 18).
2026-10-24 16:32:11 +02:00
jschoubben d6cef5829b Have a want name its composition and a backup answer; give the node its own want, the seventh rule, the held phase and the backup and record asks (issue 1943, S3)
The node agent applies only the composition its want names, and the assigner writes a want only once that
composition exists (ADR 0539 clauses 1, 6, 14 and 16).
2026-10-24 16:29:24 +02:00
jschoubben 826fd73cf2 Give a composition its value, its canonical digest and its files by digest; key a dotted module escaped (issue 1943, S2; issue 1150)
A node applies one module's composition at a time, so each carries only its own module's resources, checked
resource by resource as the node-engine reads them, and its digest follows its content alone (ADR 0539 clauses 1
and 4). The fixtures are the controller's plan answers for ace and g14, their sealed resources taken out.
2026-10-24 16:26:32 +02:00
jschoubben aee971c198 Name a node's composition bucket, its keys, and the digest with the move's judgement beside it (issue 1943, S1)
The controller writes one module's composition per node and fingerprint, and the assigner reads its digest and
judgement before it writes a want (ADR 0539 clauses 1, 2 and 14).
2026-10-24 16:21:02 +02:00
36 changed files with 3081 additions and 354 deletions
+12 -3
View File
@@ -42,11 +42,20 @@ func NodeCapacity(path, fallback, node string, logf func(string, ...any)) func()
}
setting, _ = s[CapacitySetting].(string)
}
caps, err := buildrun.ParseCapacity(setting)
n, err := capacityOf(setting, node)
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 (&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
View File
@@ -9,6 +9,7 @@ import (
"sync"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -64,10 +65,17 @@ type Puller struct {
PullWait time.Duration
Heartbeat time.Duration
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
// node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with Put, so a pause outlives
// a restart.
// 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
// reads the node's pause key (delivery.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a
// pause outlives a restart.
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
inHand int
@@ -75,6 +83,9 @@ type Puller struct {
paused bool
wake chan struct{}
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() {
@@ -113,7 +124,11 @@ func (p *Puller) setPaused(paused bool, by string) error {
if err != nil {
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)
}
}
@@ -186,13 +201,29 @@ func (p *Puller) Running() []RunningAsk {
// room says whether the puller may take one more ask now.
func (p *Puller) room() bool {
capacity := 0
if p.Capacity != nil {
ok, _, _ := p.roomNow()
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()
}
p.mu.Lock()
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
@@ -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")
}
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
if wait <= 0 {
wait = DefaultPullWait
@@ -213,12 +247,17 @@ func (p *Puller) Pull(ctx context.Context) error {
if err := ctx.Err(); err != nil {
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 {
case <-ctx.Done():
return ctx.Err()
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
}
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)
}
}
// 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, ""
}
+39
View File
@@ -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
View File
@@ -3,11 +3,15 @@ package buildstep
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"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
}
// 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) {
src, st := NewFakeAsks(), &kv{}
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
var mu sync.Mutex
capacity := 0
p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
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)
time.Sleep(50 * time.Millisecond)
if src.Waiting(delivery.AskDelivery) != 1 {
t.Fatal("a node at capacity 0 took an ask")
t.Run("settings file", func(t *testing.T) {
src, st := NewFakeAsks(), &kv{}
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
var mu sync.Mutex
capacity := 0
p := &Puller{Source: src, Language: delivery.AskGo, Node: "novox", Put: st.put, PullWait: 10 * time.Millisecond,
Capacity: func() int { mu.Lock(); defer mu.Unlock(); return capacity },
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)
time.Sleep(50 * time.Millisecond)
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
// stays paused until resumed.
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 }
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond,
Capacity: func() int { return 2 }, Run: run}
p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
if err := p.Pause("the operator"); err != nil {
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" {
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.
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Load: st.load, PullWait: 10 * time.Millisecond,
Capacity: func() int { return 2 }, Run: run}
again := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: steps.put, PausePut: st.put, Load: st.load,
PullWait: 10 * time.Millisecond, Capacity: func() int { return 2 }, Run: run}
_ = src.PublishAsk(testAsk(delivery.AskDelivery))
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
@@ -228,6 +284,27 @@ func TestPullPauseOutlivesARestart(t *testing.T) {
t.Fatal(err)
}
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
+1 -5
View File
@@ -14,11 +14,7 @@ func SeatTools(seat, what string, p *Puller) []stdio.Tool {
Run: func(map[string]any) (any, error) { return run() }}
}
current := func() (any, error) {
capacity := 0
if p.Capacity != nil {
capacity = p.Capacity()
}
return map[string]any{"node": p.Node, "capacity": capacity, "running": p.Running(), "paused": p.Paused()}, nil
return map[string]any{"node": p.Node, "capacity": p.CapacityNow(), "running": p.Running(), "paused": p.Paused()}, nil
}
return []stdio.Tool{
verb("current", "The "+what+" asks this node runs now (each one's id, kind and when it started), its build capacity, "+
+25 -46
View File
@@ -9,7 +9,26 @@ package catalogue
// DeliverySeat is the seat mesh-delivery holds.
const DeliverySeat = "mesh-delivery"
// deliveryVerbs are the delivery seat's tools: six that read, and the acts of a person, a reviewer and healer H2.
// RemovedDeliveryVerbs are the delivery seat's verbs release 6 removed (novox/hq ADR 0539 clause 17): with no plan the
// gate keeps the merge's states and derives nothing after the merge. Each stays in deliveryVerbs as an optional row
// with no input, so the holder of before and the holder of release 6 both hold the seat and a caller is answered by
// name with its replacement, given here.
var RemovedDeliveryVerbs = map[string]string{
"times": "the delivery view (nox-delivery): each machine's want and have, and since when",
"what-if": "the merge check's composition, which composes every machine a change touches",
"release": "the assigner's asks (hold, restore, backup, record), answered on the phone or in the delivery view",
"stop": "a bad fingerprint is held and its canary set back by the assigner (ADR 0429 §6, ADR 0539 clause 18)",
"retire-history": "nothing to retire: the gate keeps no delivery after its merge",
}
// deliveryRemovedInR6 is the row of a verb release 6 removed from the delivery seat.
func deliveryRemovedInR6(name string) Verb {
return Verb{Name: name, Description: "Removed in R6 (ADR 0539): refused by name, naming its replacement: " +
RemovedDeliveryVerbs[name], Input: schema(nil, nil), Optional: true}
}
// deliveryVerbs are the delivery seat's tools: what reads the merges and checks, and the acts of a person and healer H2
// on a merge (ADR 0539 clause 17).
func deliveryVerbs() []Verb {
return []Verb{
{Name: "deliveries", Description: "Every delivery not final, and those that ended in the last day, one line " +
@@ -18,15 +37,7 @@ func deliveryVerbs() []Verb {
Input: schema(map[string]string{"state": "one state, e.g. delivering or held",
"repository": "owner/repository", "group": "a group's id (its branch name)",
"all": "\"true\": the final ones of the last thirty days too"}, nil, "all")},
// Delivery time against the delivery budgets (novox/hq ADR 0282): what probe D16 reads, optional until its
// holder serves it.
{Name: "times", Description: "Delivery time: from a merge to every machine of its walk running the build. " +
"Each class's delivery budget (a leaf module three minutes, a core module ten), the last days' median and " +
"worst, how many within and over, the newest delivery of each class, every delivery over its budget with its " +
"longest phase, and the delivered ones whose time is not known. Given a delivery's id, its time phase by phase.",
Input: schema(map[string]string{"days": "how many days back (default 7)", "id": "one delivery's id: its phases"}, nil),
// Optional until mesh-delivery serves it: its holder lives in the catalogue (design 33 §7).
Optional: true},
deliveryRemovedInR6("times"),
{Name: "show", Description: "One delivery or group whole: its delivery plan (what it builds, what each " +
"machine receives, what is not an ordinary send), every transition with when and why, the machine " +
"steps of its walk, its group and its order.",
@@ -35,12 +46,7 @@ func deliveryVerbs() []Verb {
"(declared, built by, version skew, engine before controller, by name), its composed check and its " +
"state, derived from its members.",
Input: schema(map[string]string{"all": "\"true\": the groups that ended too"}, nil, "all")},
{Name: "what-if", Description: "The delivery plan a change would have, asked of the controller's planner " +
"and kept nowhere: the modules it moves and their dependents in tiers, what each machine would receive, " +
"the steps that are not an ordinary send.",
Input: schema(map[string]string{"repository": "owner/repository",
"paths": "the files it changes, comma-separated, from the repository's root",
"base": "the branch it merges into (default main)"}, []string{"repository", "paths"})},
deliveryRemovedInR6("what-if"),
{Name: "checks", Description: "What the mesh's checks said of a pull request's head or of one commit: each " +
"of the commit's mesh statuses (mesh/check, the one a main requires from R4e, hq's mesh/repo-check, " +
"mesh/delivery, …) with its state, " +
@@ -62,40 +68,13 @@ func deliveryVerbs() []Verb {
"controller is asked for its check. With why.",
Input: schema(map[string]string{"id": "the delivery's id", "why": "why, kept with the transition"},
[]string{"id", "why"})},
{Name: "release", Description: "A person's word that a held delivery goes on: it starts delivering. With why.",
Input: schema(map[string]string{"id": "the delivery's id", "why": "why, kept with the transition"},
[]string{"id", "why"})},
{Name: "stop", Description: "Stop a delivery that is not final, with why; its walk is ended through the " +
"controller, and in a group every member after it is stopped too, naming it.",
Input: schema(map[string]string{"id": "the delivery's id", "why": "why, kept with the transition"},
[]string{"id", "why"})},
deliveryRemovedInR6("release"),
deliveryRemovedInR6("stop"),
{Name: "close", Description: "Healer H2's verb: take the one transition the table names for a delivery held " +
"past its bound. Refused for any other.",
Input: schema(map[string]string{"id": "the delivery's id", "why": "the condition it answers"},
[]string{"id", "why"})},
{Name: "retire-history", Description: "Retire held deliveries that are history, on a person's word, with " +
"why: each made from the forge's word for a merge on the trunk heard long after it happened — replayed " +
"when the holder first held the seat — with no walk of it open. Never released, delivered or put back: " +
"retired, final. One by id, refused when the rule does not take it; or every held one, of one " +
"repository when named, answering how many it retired and naming those it left held. dry \"true\" " +
"answers what it would do and changes nothing.",
Input: schema(map[string]string{"why": "why, kept with every transition",
"id": "one delivery's id; every held one when empty",
"repository": "owner/repository: only its held deliveries",
"dry": "\"true\": what it would retire, changing nothing"}, []string{"why"}, "dry"),
// Added after the seat's holder shipped (novox/hq issue 313): optional until mesh-delivery serves it.
Optional: true},
{Name: "mark-ready", Description: "A reviewer's word that a delivery's head is approved: the holder puts the " +
"label mesh/ready on its pull request and then writes \"ready at <head>\", the order the forge counts, " +
"as the mesh's own forge account, naming who reviewed it and why. Refused when the delivery's commit is " +
"no longer its pull request's head. mesh-delivery then merges it once its checks pass (novox/hq ADR 0364).",
Input: schema(map[string]string{"id": "the delivery's id (owner/repository@commit): the head reviewed",
"reviewed-by": "who reviewed it, as who@node", "why": "the review's verdict in a line"},
[]string{"id", "reviewed-by", "why"}),
// Kept one more release, optional (novox/hq issue 1022): mesh-delivery no longer serves it, but the manifest
// it still has on the mesh does until its catalogue member merges, and a seat refuses a holder that serves
// a verb it does not promise. The next release removes this entry.
Optional: true},
deliveryRemovedInR6("retire-history"),
}
}
-104
View File
@@ -58,107 +58,3 @@ func TestTheDeliverySeatPromisesChecks(t *testing.T) {
t.Fatalf("a holder without %s: %v", before[0], err)
}
}
// The seat retires history on a person's word (novox/hq issue 313): `retire-history`, with why, one delivery by
// id or every held one, dry to see first. Added after the holder shipped, it is optional, so the holder that
// does not serve it yet still holds the seat, and one that does is not refused for a verb the seat lacks.
func TestTheDeliverySeatPromisesRetireHistoryOptionally(t *testing.T) {
seat, _ := SeatNamed(DeliverySeat)
var retire *Verb
for i := range seat.Serves {
if seat.Serves[i].Name == "retire-history" {
retire = &seat.Serves[i]
}
}
if retire == nil || !retire.Optional {
t.Fatalf("the delivery seat promises %v, retire-history optionally", VerbNames(seat.Serves))
}
props, _ := retire.Input["properties"].(map[string]any)
for _, arg := range []string{"why", "id", "repository", "dry"} {
if _, has := props[arg]; !has {
t.Errorf("retire-history takes no %q", arg)
}
}
if req, _ := retire.Input["required"].([]string); !reflect.DeepEqual(req, []string{"why"}) {
t.Errorf("retire-history requires %v, wanted only why", req)
}
var without []string
for _, v := range VerbNames(seat.Serves) {
if v != "retire-history" {
without = append(without, v)
}
}
m := Manifest{Module: "mesh-delivery", Claims: []Claim{{Name: DeliverySeat, Scope: ScopeMesh, Serves: without}}}
if err := CanHold(m, seat); err != nil {
t.Fatalf("a holder without retire-history yet: %v", err)
}
m.Claims[0].Serves = VerbNames(seat.Serves)
if err := CanHold(m, seat); err != nil {
t.Fatalf("a holder serving retire-history: %v", err)
}
}
// The seat says delivery times (novox/hq ADR 0282, issue 382): `times`, over some days or for one delivery. Added
// after the holder shipped, it is optional, so the holder that does not serve it yet still holds the seat, and one
// that does is not refused for a verb the seat lacks.
func TestTheDeliverySeatPromisesTimesOptionally(t *testing.T) {
seat, _ := SeatNamed(DeliverySeat)
var times *Verb
for i := range seat.Serves {
if seat.Serves[i].Name == "times" {
times = &seat.Serves[i]
}
}
if times == nil || !times.Optional {
t.Fatalf("the delivery seat promises %v, times optionally", VerbNames(seat.Serves))
}
props, _ := times.Input["properties"].(map[string]any)
for _, arg := range []string{"days", "id"} {
if _, has := props[arg]; !has {
t.Errorf("times takes no %q", arg)
}
}
var without []string
for _, v := range VerbNames(seat.Serves) {
if v != "times" {
without = append(without, v)
}
}
m := Manifest{Module: "mesh-delivery", Claims: []Claim{{Name: DeliverySeat, Scope: ScopeMesh, Serves: without}}}
if err := CanHold(m, seat); err != nil {
t.Fatalf("a holder without times yet: %v", err)
}
m.Claims[0].Serves = VerbNames(seat.Serves)
if err := CanHold(m, seat); err != nil {
t.Fatalf("a holder serving times: %v", err)
}
}
// The seat still promises `mark-ready`, optionally, for one more release (issue 1022); it was added by novox/hq issue 913: `mark-ready`, with the delivery's id, who reviewed it
// and why, so a reviewer marks a pull request through one audited verb, label and comment in the order the forge
// counts, instead of two raw forge calls. Added after the holder shipped, it is optional.
func TestTheDeliverySeatPromisesMarkReadyOptionally(t *testing.T) {
seat, _ := SeatNamed(DeliverySeat)
var mark *Verb
for i := range seat.Serves {
if seat.Serves[i].Name == "mark-ready" {
mark = &seat.Serves[i]
}
}
if mark == nil || !mark.Optional {
t.Fatalf("the delivery seat promises %v, mark-ready optionally", VerbNames(seat.Serves))
}
if req, _ := mark.Input["required"].([]string); !reflect.DeepEqual(req, []string{"id", "reviewed-by", "why"}) {
t.Errorf("mark-ready requires %v, wanted id, reviewed-by and why", req)
}
var without []string
for _, v := range VerbNames(seat.Serves) {
if v != "mark-ready" {
without = append(without, v)
}
}
m := Manifest{Module: "mesh-delivery", Claims: []Claim{{Name: DeliverySeat, Scope: ScopeMesh, Serves: without}}}
if err := CanHold(m, seat); err != nil {
t.Fatalf("a holder without mark-ready yet: %v", err)
}
}
+6 -3
View File
@@ -110,15 +110,18 @@ var defaultSeats = append([]Seat{
"rolled-back",
// A pull request's merge check, judged (novox/hq to-be 45 §9).
"checked",
// A walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
// A walk kept in a new state, and a tier sent to a machine: kept until the catalogue's main consumes them no
// more; nothing emits them (ADR 0539).
"plan-moved",
// A tier sent to a machine, once the send is kept (novox/hq ADR 0406 §5, issue 974).
"sent",
// A key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
"assignment-changed", "declared-config-changed", "predefined-changed",
// A builder's key not taken in because the artifact store does not hold an artifact it names (novox/hq
// issue 1937): the builder builds it again.
"artifacts-gone"},
"artifacts-gone",
// A bucket of a state this release deleted, deleted at the controller's first start (novox/hq ADR 0539
// clause 13): the merge window's, the silent sends' and the building plans' hold, each said once.
"bucket-deleted"},
Serves: ControllerVerbs},
// The store's first verbs (novox/hq ADR 0159): the smallest set that makes the store askable,
// served by whichever module holds the seat with tools of these names.
+46 -106
View File
@@ -74,6 +74,29 @@ func VerbNames(verbs []Verb) []string {
return out
}
// RemovedControllerVerbs are the controller seat's verbs release 6 removed (novox/hq ADR 0539, ADR 0429 §8): no
// delivery plan, batch, tier or push exists, so nothing starts, stops, retries, follows or reads one. Each stays in
// ControllerVerbs as a row with no input, so a caller is answered by name with its replacement, given here.
var RemovedControllerVerbs = map[string]string{
"plans": "the delivery view (nox-delivery): each machine's want and have, read from the assigner's wants and the " +
"node-engine's haves",
"plan": "the node's compositions: mesh-controller_composed holds each module's digest and judgement per node, and " +
"mesh-hand-compose prints one module's composition for one node",
"push": "a want: the assigner writes want.<node>.<module> by its rules, and by hand the operator's credential writes " +
"it, naming a kept composition (ADR 0539 clause 10)",
"deliver": "nothing to start: a merged change reaches each machine by its want",
"delivery-stop": "a bad fingerprint is held and its canary set back by the assigner (ADR 0429 §6, ADR 0539 clause 18)",
"delivery-backups-again": "the assigner's backup ask, answered on the phone or in the delivery view (ADR 0539 clause 16)",
"delivery-walks": "the delivery view (nox-delivery) and the wants, haves and compositions on the bus",
"delivery-order": "no order to keep: each module reaches its machines by its own want, its providers first by rule 5",
}
// removedInR6 is the row of a verb release 6 removed from the controller's seat: no input, refused by name.
func removedInR6(name string) Verb {
return Verb{Name: name, Description: "Removed in R6 (ADR 0539): refused by name, naming its replacement: " +
RemovedControllerVerbs[name], Input: schema(nil, nil)}
}
// ControllerSeatName is the seat the control plane holds, whose tools are the mesh's own verbs.
const ControllerSeatName = "mesh-controller"
@@ -115,87 +138,23 @@ var ControllerVerbs = []Verb{
"log": "a build's id (as `builds` lists it): print what the build machine said, line by line",
"limit": "how many builds to list (default 20); not with log",
}, nil)},
{Name: "plans", Description: "What the last merges produced and where each stands (novox/hq ADR 0162): " +
"the tiers, the tier a plan is at, what it waits for and since when; one plan whole, given its id; " +
"or, given wait, one plan followed until it moves.",
Input: schema(map[string]string{
"id": "a plan's id (as `plans` lists them): that plan, tier by tier",
"wait": "a plan's id: wait until it moves, then answer it as data — the plan, and what ended the wait " +
"(moved, tier, done, or bound); ask again to follow it further (novox/hq issue 229)",
"until": "with wait: moved (any change, the default), tier (it reaches another tier) or done (it ends)",
"bound": "with wait: how long to wait at most, as 15s or seconds — 20s, the default, is the longest, " +
"inside what a call is answered within",
"stop": "a plan's id: stop it — what was asked still builds, nothing further is asked",
"close": "a plan's id: close a plan that will not move again, as failed by hand (novox/hq issue 254)",
"retry": "a failed plan's id: ask its failed builds again under new ids, and carry the plan on from that tier (novox/hq ADR 0219)",
"go": "a walk's id that waits for its delivery's word: start it by hand, in place of mesh-delivery — with why, " +
"recorded in the hand-act log (novox/hq ADR 0239)",
"repository": "owner/repository: the plan a merge there would produce, saving nothing (what-if); with paths or modules",
"paths": "with repository: the files the merge would change, comma-separated, from the repository's root",
"modules": "with repository: or the modules it would change, comma-separated",
"module-dirs": "with repository and paths: the directories holding a module.json at the commit, comma-separated, " +
"as the forge's announcer says them (novox/hq issue 278); without it, a directory is a module only when the paths hold its manifest",
"data": "with repository and paths: true answers the change plan as data, the document delivery-plan answers " +
"(moved, dependents, new, tiers, machines with what each receives and waits for, steps, summary, class, head, base); " +
"without it the answer is text (novox/hq ADR 0450 §2)",
"head": "with data: the commit at hand",
"base": "with data: the branch it merges into (default main)",
"removed": "with data: the files among paths the change deletes, comma-separated",
"limit": "how many plans to list (default 10); only when listing",
"why": "with stop, close or go: why it is done by hand — required, and recorded in the hand-act log (novox/hq to-be 45 §7)",
"cause": "with stop, close or go: the cause in a word, or a condition's kind (optional)",
}, nil, "data")},
removedInR6("plans"),
// The verbs the delivery's owner asks with (novox/hq ADR 0239): the planner's answer, a group's order and
// composed check, and the start and end of a walk. Callable by a person too; mesh-delivery is who needs them.
{Name: "delivery-plan", Description: "Removed in R4e (ADR 0501 §5 / ADR 0503 §3): refused by name, naming its replacement",
Input: schema(nil, nil)},
{Name: "delivery-order", Description: "A delivery group's order (novox/hq ADR 0419): its members in the order " +
"they are delivered, every pair and why — built by, verb before caller, version skew, engine before " +
"controller — read from the module graph and the manifests' change alone; no `after:` line is read. The " +
"cycle, with every module and edge of it, is the one refusal.",
Input: schema(map[string]string{"members": "the members, as JSON: [{id, repository, base, head, number, paths, " +
"module_dirs, module_dirs_said, removed}]"}, []string{"members"})},
removedInR6("delivery-order"),
{Name: "delivery-check", Description: "Answers that the checker (mesh-delivery-checker) judges a group (novox/hq " +
"ADR 0503 §3); asks nothing. Its inputs stay, since hq's repository check still names it (ADR 0513 §4).",
Input: schema(map[string]string{"group": "the group's id (none for one head checked again)",
"members": "the members, as JSON, as delivery-order takes them",
"ready": "\"true\" when the pull request is marked ready: its check is queued before every other check (novox/hq ADR 0406 §3)"},
[]string{"members"})},
{Name: "deliver", Description: "The delivery's word that a walk waiting for it may start (novox/hq ADR 0239): its " +
"first tier is asked at the controller's next pass. Refused for a walk that waits for nobody.",
Input: schema(map[string]string{"plan": "the walk's id", "why": "what lets it go: its turn, a person's release"},
[]string{"plan"})},
{Name: "delivery-stop", Description: "End a walk on its delivery's word (novox/hq ADR 0239): failed, said as stopped " +
"by whom and why; what it asked still builds and registers, nothing further is asked or sent. Given merges to " +
"leave out, every other merge the walk answered is walked again — in the open batch of its kind, or a new one — " +
"without them, without each later merge of their repository, which contains one, and without each merge moving " +
"a module one moves, each left out too and named (novox/hq ADR 0299); a batch not yet cut drops the merges. The " +
"answer names the batch, the merges walked again, the merges left out, the modules the stopped walk sent that " +
"stay on their machines, and the merges elsewhere that will still deliver a stopped change.",
Input: schema(map[string]string{"plan": "the walk's id", "why": "why", "by": "who stopped the delivery",
"without": "merges to leave out, <repository>@<commit>, comma-separated"},
[]string{"plan", "why"})},
{Name: "delivery-backups-again", Description: "Ask again the backups a walk was held for (novox/hq ADR 0404 " +
"decision 4): a walk whose backup before its send failed or did not answer is held, nothing asked or sent; this " +
"clears those failed backups and the next look at the plans asks them again, and only them. Refused for a walk " +
"with no failed backup. The delivery's release calls it. The answer names the walk and the backups asked again.",
Input: schema(map[string]string{"plan": "the held walk's id", "why": "a person's reason, as release passes it",
"by": "who released the delivery"}, []string{"plan", "why"})},
{Name: "delivery-walks", Description: "The walks the controller keeps (novox/hq ADR 0239): every open one and the " +
"last ended ones, each whole — its tiers, each module's state, first machines and first-node gate with its readings, and its " +
"phases from its merge to every machine running it (novox/hq ADR 0282) — and whether the delivery seat has a " +
"holder on record. Given a plan, that one.",
Input: schema(map[string]string{"plan": "one walk's id", "limit": "how many ended walks beside the open ones (default 50)"},
nil)},
{Name: "plan", Description: "What one machine would run, and why: the declaration the mesh would send it — " +
"or, with files, the files it would be given; or, with diff, what a push would change there against what " +
"it was last sent: resources added, removed and changed, a container's changed fields (it is recreated), " +
"and any mount whose directory on the machine would move (novox/hq ADR 0217).",
Input: schema(map[string]string{
"node": "the machine's name",
"files": "\"true\": the files this machine would be given, instead of the declaration as JSON",
"diff": "\"true\": what a push would change on this machine, against what it was last sent; not with files",
}, []string{"node"}, "files", "diff")},
removedInR6("deliver"),
removedInR6("delivery-stop"),
removedInR6("delivery-backups-again"),
removedInR6("delivery-walks"),
removedInR6("plan"),
{Name: "assign", Description: "Put a module on a machine. Refused with the mesh's own words when it cannot resolve there, " +
"or when a seat its resources are applied through is held by nothing on the machine (novox/hq ADR 0207). " +
"A module not registered yet is looked for (novox/hq issue 325, ADR 0261): while its build runs — a merge " +
@@ -221,22 +180,7 @@ var ControllerVerbs = []Verb{
}, []string{"node", "provision", "from", "module"})},
{Name: "unpin", Description: "Take that choice back, putting the question to the mesh again.",
Input: schema(map[string]string{"node": "the machine's name", "provision": "the provision"}, []string{"node", "provision"})},
{Name: "push", Description: "Send one machine everything it should be. With no machine named it is a push of the " +
"WHOLE mesh — every machine that is behind — and the answer says so first; behind says that outright. " +
"Answers at once that it is running, with a call id: `calls` with that id says what it sent and, for a " +
"machine named, how that machine applied it — applied, refused with its reason, or each failed resource — " +
"once it reports within wait (a push can reload the bus, which then refuses any answer still to come). A push by hand is a repair, " +
"and says why: recorded in the hand-act log (novox/hq to-be 45 §7). A machine whose push would give a " +
"running module's data another directory is held and named, the others sent (novox/hq ADR 0217): read " +
"`plan` with diff, and name the machine with move to send it.",
Input: schema(map[string]string{
"node": "the machine's name; without it, every machine that is behind",
"move": "with node: the modules whose data may move with this push, comma-separated (optional)",
"wait": "with node: how long the call waits for the machine to report how it applied what it was sent, as \"90s\" or \"2m\"; \"0\" waits for nothing; without it, the controller's setting push-wait (optional)",
"behind": "\"true\": every machine that is behind, the whole mesh — the same as naming none, said outright; not with node",
"why": "why this is pushed by hand: recorded in the hand-act log",
"cause": "the cause in a word, or a condition's kind — the word a second push for the same reason uses (optional)",
}, []string{"why"}, "behind")},
removedInR6("push"),
{Name: "rotate", Description: "Replace a credential. A pair credential, by provision (and a consuming machine and module, " +
"else every holder): both ends are re-sent together. Or a module's own secret, by machine, module and " +
"name: made anew and the machine sent, so the module starts again on it — for a secret its definition " +
@@ -457,26 +401,22 @@ var ControllerVerbs = []Verb{
"probe": "one probe's id (D14): its verdict alone, with every finding — the verdict shows at most twenty a probe",
}, nil, "run", "probes", "signals")},
// How a module's new builds reach its machines, and the bus's planned step (novox/hq ADR 0236).
{Name: "upgrade", Description: "How each module's new builds reach its machines (novox/hq ADR 0236): rolled " +
"out one machine first and judged there at the gate, then the rest — or recorded, waiting for a person's " +
"push — with where that comes from (a person, the module, the bus, its irreplaceable data, the default) and " +
"why. With module, that one; with policy, a person's choice for it — roll-out, record (with why) or default " +
"to take the choice back. The bus is never rolled out.",
{Name: "upgrade", Description: "Whether each module's new builds wait for a person (novox/hq ADR 0236, ADR 0539 clause 14): " +
"for a module of policy record, the assigner writes no want naming a composition judged to recreate a container or move " +
"data until the operator answers its ask; with where the policy comes from (a person, the module, the bus, its " +
"irreplaceable data, the default) and why. With module, that one; with policy, a person's choice for it — roll-out, " +
"record (with why) or default to take the choice back. The bus is never moved by the assigner (ADR 0539 clause 15).",
Input: schema(map[string]string{
"module": "one module",
"policy": "with module: roll-out, record or default",
"together": "\"true\": with roll-out, every machine at once instead of one machine first",
"why": "with policy or release-backlog: why — required for record and for a release, kept",
"backlog": "\"true\": every build waiting for a gate, per machine — what a release plan would send",
"release-backlog": "\"true\": release what waits now, one machine at a time, each judged — after a release " +
"plan failed, the mesh waits for this; with why",
"cause": "with release-backlog: the cause in a word (optional)",
}, nil, "together", "backlog", "release-backlog")},
{Name: "bus", Description: "The bus as a planned step (novox/hq to-be 45 §8, ADR 0236): what a bus upgrade " +
"would do — the bus's build on each machine against the one the mesh holds — and how the last step went. " +
"With upgrade, start one: a person's act with why, after the streams are snapshotted (snapshot-taken says " +
"where, while the mesh takes none itself), saying first whether it can be reverted; bus-maintenance is open " +
"while it runs and every stream, consumer and a round trip are checked after.",
"module": "one module",
"policy": "with module: roll-out, record or default",
"why": "with policy: why — required for record, kept",
}, nil)},
{Name: "bus", Description: "The bus as a person's step (novox/hq to-be 45 §8, ADR 0236, ADR 0539 clause 15): what a bus " +
"upgrade would do — the bus's build on each machine against the one the mesh holds — and how the last step went. " +
"With upgrade, check one: a person's act with why, after the streams are snapshotted (snapshot-taken says where, " +
"while the mesh takes none itself), saying first whether it can be reverted; then it answers the want the operator " +
"writes with the operator's credential, want.<node>.<bus module>, recorded — nothing is sent, and the assigner never " +
"moves the bus.",
Input: schema(map[string]string{
"upgrade": "\"true\": start the bus's upgrade",
"why": "with upgrade: why — required, recorded in the hand-act log",
+115
View File
@@ -0,0 +1,115 @@
package catalogue
import (
"strings"
"testing"
)
// 1943/AC4: the controller's seat no longer offers a verb that starts, stops, retries, follows or reads a delivery
// plan, nor a push: each is a row with no input whose description refuses it by name and names its replacement, and
// each is in the removed list the holder answers from. upgrade keeps its policy and drops the plan's backlog.
func TestTheControllerSeatHasNoPlanVerbs(t *testing.T) {
rows := map[string]Verb{}
for _, v := range ControllerVerbs {
rows[v.Name] = v
}
gone := []string{"plans", "plan", "push", "deliver", "delivery-stop", "delivery-backups-again", "delivery-walks", "delivery-order"}
for _, name := range gone {
v, ok := rows[name]
if !ok {
t.Errorf("%s is not in the table, so a caller is not answered by name", name)
continue
}
replacement, removed := RemovedControllerVerbs[name]
if !removed || replacement == "" {
t.Errorf("%s is not in the removed list with its replacement", name)
}
if !strings.HasPrefix(v.Description, "Removed in R6 (ADR 0539): refused by name") || !strings.Contains(v.Description, replacement) {
t.Errorf("%s = %q, want refused by name, naming %q", name, v.Description, replacement)
}
if props, _ := v.Input["properties"].(map[string]any); len(props) > 0 {
t.Errorf("%s still takes %v", name, props)
}
}
props, _ := rows["upgrade"].Input["properties"].(map[string]any)
for _, arg := range []string{"together", "backlog", "release-backlog", "cause"} {
if _, has := props[arg]; has {
t.Errorf("upgrade still takes %s, a plan's", arg)
}
}
for _, arg := range []string{"module", "policy", "why"} {
if _, has := props[arg]; !has {
t.Errorf("upgrade no longer takes %s", arg)
}
}
for _, word := range []string{"rolled", "gate", "backlog", "push"} {
if strings.Contains(rows["upgrade"].Description, word) {
t.Errorf("upgrade's description still speaks of %q: %q", word, rows["upgrade"].Description)
}
}
if !strings.Contains(rows["bus"].Description, "want") {
t.Errorf("bus does not say the operator writes the bus's want: %q", rows["bus"].Description)
}
for name := range RemovedControllerVerbs {
if v := rows[name]; !strings.HasPrefix(v.Description, "Removed in R6") {
t.Errorf("%s is in the removed list and its row says %q", name, v.Description)
}
}
}
// 1943/AC4: the delivery seat keeps the merge's verbs and no longer offers times, what-if, release, stop or
// retire-history: each is a row with no input, optional so the holder of before and the holder of release 6 both hold
// the seat, refusing it by name; mark-ready, served by nobody since issue 1022, is gone.
func TestTheDeliverySeatHasNoPlanVerbs(t *testing.T) {
seat, ok := SeatNamed(DeliverySeat)
if !ok {
t.Fatal("the delivery seat is not in the set")
}
rows := map[string]Verb{}
for _, v := range seat.Serves {
rows[v.Name] = v
}
for _, name := range []string{"times", "what-if", "release", "stop", "retire-history"} {
v, ok := rows[name]
replacement, removed := RemovedDeliveryVerbs[name]
if !ok || !removed || replacement == "" {
t.Errorf("%s: in the table %v, in the removed list %v", name, ok, removed)
continue
}
if !strings.HasPrefix(v.Description, "Removed in R6 (ADR 0539): refused by name") || !strings.Contains(v.Description, replacement) || !v.Optional {
t.Errorf("%s = %q (optional %v), want an optional row refusing it by name", name, v.Description, v.Optional)
}
if props, _ := v.Input["properties"].(map[string]any); len(props) > 0 {
t.Errorf("%s still takes %v", name, props)
}
}
if _, ok := rows["mark-ready"]; ok {
t.Errorf("mark-ready is still promised")
}
for _, kept := range []string{"deliveries", "show", "groups", "checks", "table", "recheck", "close", "stalled"} {
if v, ok := rows[kept]; !ok || strings.HasPrefix(v.Description, "Removed") {
t.Errorf("%s is not kept (ADR 0539 clause 17)", kept)
}
}
gate := Manifest{Module: "mesh-delivery", Claims: []Claim{{Name: DeliverySeat, Scope: ScopeMesh,
Serves: []string{"deliveries", "show", "groups", "checks", "table", "stalled", "recheck", "close"}}}}
if err := CanHold(gate, seat); err != nil {
t.Errorf("the gate of release 6 cannot hold the seat: %v", err)
}
gate.Claims[0].Serves = append(gate.Claims[0].Serves, "times", "what-if", "release", "stop", "retire-history")
if err := CanHold(gate, seat); err != nil {
t.Errorf("the gate of release 5 cannot hold the seat: %v", err)
}
}
// 1943 (for 1945/AC14): the controller's seat says a bucket it deleted at its first start as its event, bucket-deleted
// (ADR 0539 clause 13).
func TestTheControllerSaysABucketItDeleted(t *testing.T) {
seat, ok := SeatNamed(ControllerSeatName)
if !ok {
t.Fatal("no controller seat")
}
if !SeatSays(seat.Emits, "bucket-deleted") {
t.Errorf("the controller's seat does not emit bucket-deleted: %v", seat.Emits)
}
}
+209
View File
@@ -0,0 +1,209 @@
// mesh-hand-compose composes one module for one node by hand, from the inputs the controller keeps, and signs it with the
// operator's key, for the hand path of novox/hq ADR 0539 clause 10: a fingerprint the controller never composed (a
// hand build), while the controller cannot compose its own fix. It composes with mesh-sdk's composition, the one the
// controller writes with (compose.Compositions), and prints the composition, its key and its digest; only with -write
// does it write them, with the operator's credential: the composition into the node's bucket composed-<node>, each
// file over 64 KiB into the node's object bucket, and the digest into mesh-controller_composed. The want is the
// operator's next act, naming that digest; each act is recorded in the hand-act log.
//
// mesh-hand-compose -input <inputs.json> -node <node> -module <module> -fp <src2:...> -key <operator key>
// [-write -nats <url> -creds <operator credential>]
//
// The inputs are a compose.Input as JSON. The key is the operator's ed25519 private key, PKCS#8 in PEM; the
// node-engine trusts it beside the controller's (ADR 0539 clause 3).
package main
import (
"context"
"crypto/ed25519"
"crypto/x509"
"encoding/json"
"encoding/pem"
"errors"
"flag"
"fmt"
"io"
"os"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/declaration"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func main() {
if err := run(os.Args[1:], os.Stdout); err != nil {
fmt.Fprintln(os.Stderr, "mesh-hand-compose:", err)
os.Exit(1)
}
}
// printed is what mesh-hand-compose prints: where the composition goes, its digest, the composition, and whether it was
// written.
type printed struct {
Bucket string `json:"bucket"`
Key string `json:"key"`
Digest string `json:"digest"`
Composition json.RawMessage `json:"composition"`
Objects []string `json:"objects,omitempty"`
Written bool `json:"written"`
Next string `json:"next"`
}
func run(args []string, stdout io.Writer) error {
fs := flag.NewFlagSet("mesh-hand-compose", flag.ContinueOnError)
input := fs.String("input", "", "the controller's kept inputs, a compose.Input as JSON")
node := fs.String("node", "", "the node to compose for")
module := fs.String("module", "", "the module to compose")
fp := fs.String("fp", "", "the module's fingerprint, src2: and 64 hex digits")
keyFile := fs.String("key", "", "the operator's ed25519 private key, PKCS#8 in PEM")
write := fs.Bool("write", false, "write the composition, its files and its digest, with the operator's credential")
url := fs.String("nats", nats.DefaultURL, "with -write: the bus")
creds := fs.String("creds", "", "with -write: the operator's credential file")
if err := fs.Parse(args); err != nil {
return err
}
switch {
case *input == "" || *node == "" || *module == "" || *fp == "":
return errors.New("-input, -node, -module and -fp are each required")
case *keyFile == "":
return errors.New("-key is required: a composition made by hand is signed by the operator's key, and the node-engine applies no other")
case *write && *creds == "":
return errors.New("-write needs -creds: the operator's credential is the only hand writer of a composition")
}
key, err := readKey(*keyFile)
if err != nil {
return err
}
key2, err := delivery.ComposedKey(*node, *module, *fp)
if err != nil {
return err
}
bucket, err := delivery.ComposedBucketFor(*node)
if err != nil {
return err
}
raw, err := os.ReadFile(*input)
if err != nil {
return err
}
var in compose.Input
if err := json.Unmarshal(raw, &in); err != nil {
return fmt.Errorf("the inputs in %s are not a compose.Input: %w", *input, err)
}
all, _, err := compose.Compositions(in, *node, map[string]string{*module: *fp}, compose.Reading)
if err != nil {
return err
}
c, ok := all[key2]
if !ok {
return fmt.Errorf("%s does not run %s: its composition holds none of it", *node, *module)
}
c, objects, err := c.ByDigest()
if err != nil {
return err
}
if c, err = delivery.SignComposed(c, key); err != nil {
return err
}
encoded, err := delivery.EncodeComposed(c)
if err != nil {
return err
}
digest, err := delivery.ComposedDigestOf(c)
if err != nil {
return err
}
out := printed{Bucket: bucket, Key: key2, Digest: digest, Composition: encoded,
Next: fmt.Sprintf("write want.%s.%s naming fp %s and composed %s with the operator's credential, and record each act in the hand-act log", *node, *module, *fp, digest)}
for d := range objects {
out.Objects = append(out.Objects, d)
}
if *write {
if err := writeIt(*url, *creds, *node, bucket, key2, encoded, digest, objects); err != nil {
return err
}
out.Written = true
}
enc := json.NewEncoder(stdout)
enc.SetIndent("", " ")
return enc.Encode(out)
}
func readKey(path string) (ed25519.PrivateKey, error) {
raw, err := os.ReadFile(path)
if err != nil {
return nil, err
}
block, _ := pem.Decode(raw)
if block == nil {
return nil, fmt.Errorf("%s holds no PEM block", path)
}
parsed, err := x509.ParsePKCS8PrivateKey(block.Bytes)
if err != nil {
return nil, fmt.Errorf("%s: %w", path, err)
}
key, ok := parsed.(ed25519.PrivateKey)
if !ok {
return nil, fmt.Errorf("%s is not an ed25519 key", path)
}
return key, nil
}
// writeIt writes the files by digest, then the composition, then its digest: a reader that finds the digest finds the
// composition and its files already there.
func writeIt(url, creds, node, bucket, key string, encoded []byte, digest string, objects map[string][]byte) error {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
nc, err := nats.Connect(url, nats.UserCredentials(creds), nats.Name("mesh-hand-compose"))
if err != nil {
return err
}
defer nc.Close()
js, err := jetstream.New(nc)
if err != nil {
return err
}
if len(objects) > 0 {
filesBucket, err := declaration.FileBucketFor(node)
if err != nil {
return err
}
store, err := js.ObjectStore(ctx, filesBucket)
if err != nil {
return fmt.Errorf("the node's object bucket %s: %w", filesBucket, err)
}
for d, body := range objects {
name, err := declaration.ObjectName(d)
if err != nil {
return err
}
if _, err := store.PutBytes(ctx, name, body); err != nil {
return fmt.Errorf("the file %s: %w", d, err)
}
}
}
composed, err := js.KeyValue(ctx, bucket)
if err != nil {
return fmt.Errorf("the bucket %s: %w", bucket, err)
}
if _, err := composed.Put(ctx, key, encoded); err != nil {
return fmt.Errorf("%s in %s: %w", key, bucket, err)
}
value, err := delivery.EncodeComposedDigest(delivery.ComposedDigest{Digest: digest, Judgement: delivery.JudgedCannot,
Words: "composed by hand with mesh-hand-compose (ADR 0539 clause 10): the operator writes the want"})
if err != nil {
return err
}
digests, err := js.KeyValue(ctx, delivery.ControllerComposedBucket)
if err != nil {
return fmt.Errorf("the bucket %s: %w", delivery.ControllerComposedBucket, err)
}
if _, err := digests.Put(ctx, key, value); err != nil {
return fmt.Errorf("%s in %s: %w", key, delivery.ControllerComposedBucket, err)
}
return nil
}
+145
View File
@@ -0,0 +1,145 @@
package main
import (
"bytes"
"crypto/ed25519"
"crypto/rand"
"crypto/x509"
"encoding/base64"
"encoding/json"
"encoding/pem"
"os"
"path/filepath"
"sort"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/overlay"
)
// aMesh is the Input of mesh-controller's aMesh, as mesh-sdk's compose tests make it (compose/fixture_test.go): two
// machines on the private network, the module the mesh provides for itself assigned to both. The manifest is the one
// the controller provides (overlay.Manifest), read back as the catalogue reads it.
func aMesh(t *testing.T) compose.Input {
t.Helper()
var m catalogue.Manifest
b, _ := json.Marshal(overlay.Manifest())
if err := json.Unmarshal(b, &m); err != nil {
t.Fatal(err)
}
in := compose.Input{Catalogue: map[string]catalogue.Manifest{overlay.Name: m}}
for i, name := range []string{"anchor", "laptop"} {
pub, _, _ := ed25519.GenerateKey(rand.Reader)
in.Nodes = append(in.Nodes, compose.Node{Name: name,
Place: compose.Place{Key: base64.StdEncoding.EncodeToString(pub), Endpoint: name + ".example:51820", Site: "here", Hub: name == "anchor", Address: []string{"10.77.0.1", "10.77.0.2"}[i]},
Assigned: []string{overlay.Name}, Capabilities: map[string]bool{"container-runtime": true, "wireguard": true, "systemd": true}})
}
return in
}
func writeKey(t *testing.T, dir string) ed25519.PublicKey {
t.Helper()
pub, priv, _ := ed25519.GenerateKey(rand.Reader)
der, err := x509.MarshalPKCS8PrivateKey(priv)
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dir, "operator.key"), pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: der}), 0o600); err != nil {
t.Fatal(err)
}
return pub
}
// 1943/AC5: mesh-hand-compose, given the kept inputs, a node, a module and its fingerprint, prints that module's
// composition for that node signed by the operator's key and its digest: its resources are the controller's for that
// module on that node (compose.Compose), and it verifies against the operator's key. Without --write it writes nothing.
func TestMeshComposeGivesTheControllersResourcesSignedByTheOperator(t *testing.T) {
dir := t.TempDir()
in := aMesh(t)
raw, err := json.Marshal(in)
if err != nil {
t.Fatal(err)
}
input := filepath.Join(dir, "input.json")
if err := os.WriteFile(input, raw, 0o600); err != nil {
t.Fatal(err)
}
operator := writeKey(t, dir)
fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
var out bytes.Buffer
if err := run([]string{"-input", input, "-node", "laptop", "-module", overlay.Name, "-fp", fp,
"-key", filepath.Join(dir, "operator.key")}, &out); err != nil {
t.Fatal(err)
}
var printed struct {
Key string `json:"key"`
Bucket string `json:"bucket"`
Digest string `json:"digest"`
Composition json.RawMessage `json:"composition"`
Written bool `json:"written"`
}
if err := json.Unmarshal(out.Bytes(), &printed); err != nil {
t.Fatalf("%s: %v", out.String(), err)
}
c, err := delivery.DecodeComposed(printed.Composition)
if err != nil {
t.Fatal(err)
}
if digest, err := delivery.VerifyComposed(c, []ed25519.PublicKey{operator}); err != nil || digest != printed.Digest {
t.Errorf("the composition verifies %q, %v against the operator's key; printed %s", digest, err, printed.Digest)
}
wantKey, _ := delivery.ComposedKey("laptop", overlay.Name, fp)
if printed.Key != wantKey || printed.Bucket != "composed-laptop" || printed.Written {
t.Errorf("printed key %s in %s, written %v", printed.Key, printed.Bucket, printed.Written)
}
d, _, err := compose.Compose(in, "laptop")
if err != nil {
t.Fatal(err)
}
var want, got []string
for _, r := range d.Resources {
want = append(want, r["id"].(string))
}
for _, r := range c.Resources {
var head struct{ ID string }
_ = json.Unmarshal(r, &head)
got = append(got, head.ID)
}
sort.Strings(want)
sort.Strings(got)
if len(got) == 0 || strings.Join(got, " ") != strings.Join(want, " ") {
t.Errorf("mesh-hand-compose gives %v, the controller's composition %v", got, want)
}
}
// 1943/AC5: mesh-hand-compose refuses what it cannot compose, naming why: no key, a fingerprint not of the new form, a
// module the node does not run, a node the mesh does not have.
func TestMeshComposeRefuses(t *testing.T) {
dir := t.TempDir()
raw, _ := json.Marshal(aMesh(t))
input := filepath.Join(dir, "input.json")
_ = os.WriteFile(input, raw, 0o600)
writeKey(t, dir)
key := filepath.Join(dir, "operator.key")
fp := delivery.SourcePrefix2 + strings.Repeat("a", 64)
for _, c := range []struct {
args []string
why string
}{
{[]string{"-input", input, "-node", "laptop", "-module", overlay.Name, "-fp", fp}, "-key"},
{[]string{"-input", input, "-node", "laptop", "-module", overlay.Name, "-fp", "src2:1", "-key", key}, "fingerprint"},
{[]string{"-input", input, "-node", "laptop", "-module", "mesh-mail", "-fp", fp, "-key", key}, "does not run mesh-mail"},
{[]string{"-input", input, "-node", "nowhere", "-module", overlay.Name, "-fp", fp, "-key", key}, "nowhere"},
{[]string{"-input", input, "-node", "laptop", "-module", overlay.Name, "-fp", fp, "-key", key, "-write"}, "-creds"},
} {
var out bytes.Buffer
if err := run(c.args, &out); err == nil || !strings.Contains(err.Error(), c.why) {
t.Errorf("mesh-hand-compose %v = %v, want a refusal saying %q", c.args, err, c.why)
}
}
}
+94
View File
@@ -0,0 +1,94 @@
package compose
import (
"encoding/json"
"fmt"
"sort"
"git.novox.be/novox/mesh-sdk/go/declaration"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Compositions is one node's compositions (novox/hq ADR 0539 clauses 1 and 2): its declaration composed over in, as
// the controller composes it, split into each module's composition at the fingerprint fps gives it and the node's own
// (`<node>.node`), by key (delivery.ComposedKey). Each resource says its module, whatever the node reported it reads.
// A module fps gives no fingerprint for is not composed and is named in skipped, so a caller composing one module
// (mesh-hand-compose) gets that one. Each module's Data says whether its manifest declares data. The compositions are not
// signed and not yet by digest: the writer does both (delivery.SignComposed, Composed.ByDigest).
func Compositions(in Input, node string, fps map[string]string, mode Mode) (out map[string]delivery.Composed, skipped []string, err error) {
reading := in
reading.Nodes = append([]Node(nil), in.Nodes...)
found := false
for i := range reading.Nodes {
if reading.Nodes[i].Name == node {
reading.Nodes[i].ReadsModules = true
found = true
}
}
if !found {
return nil, nil, fmt.Errorf("compose: the mesh has no node %q", node)
}
d, problems, err := ComposeIn(reading, node, mode)
if err != nil {
return nil, nil, err
}
if len(problems) > 0 {
return nil, nil, fmt.Errorf("compose: %s's declaration is refused by the node-engine's validator: %v", node, problems)
}
body, err := d.Body()
if err != nil {
return nil, nil, err
}
var envelope struct {
Resources []json.RawMessage `json:"resources"`
Adoption *declaration.Adoption `json:"adoption"`
Filled []string `json:"filled_files"`
Provisioner string `json:"provisioner_account"`
}
if err := json.Unmarshal(body, &envelope); err != nil {
return nil, nil, err
}
var kept []json.RawMessage
keptPaths := map[string]bool{}
left := map[string]bool{}
for _, raw := range envelope.Resources {
var head struct {
Module string `json:"module"`
Type string `json:"type"`
Path string `json:"path"`
}
if err := json.Unmarshal(raw, &head); err != nil {
return nil, nil, err
}
if _, ok := fps[head.Module]; head.Module != "" && !ok {
left[head.Module] = true
continue
}
if head.Type == string(declaration.TypeFile) {
keptPaths[head.Path] = true
}
kept = append(kept, raw)
}
var filled []string
for _, p := range envelope.Filled {
if keptPaths[p] {
filled = append(filled, p)
}
}
out, err = delivery.SplitComposed(node, kept, fps, filled,
delivery.NodeOwn{Adoption: envelope.Adoption, Provisioner: envelope.Provisioner})
if err != nil {
return nil, nil, err
}
for key, c := range out {
if m, ok := in.Catalogue[c.Module]; ok && c.Module != "" {
c.Data = len(m.DataItems()) > 0
out[key] = c
}
}
for m := range left {
skipped = append(skipped, m)
}
sort.Strings(skipped)
return out, skipped, nil
}
+88
View File
@@ -0,0 +1,88 @@
package compose
import (
"encoding/json"
"sort"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
func fpFor(c byte) string { return delivery.SourcePrefix2 + strings.Repeat(string(c), 64) }
// 1943/AC5: a node's compositions are its declaration as Compose gives it, split by the module each resource belongs
// to: every resource once, each in its module's composition at the fingerprint given, the module's ownership said on
// it; a module given no fingerprint is named as skipped and composed nowhere.
func TestCompositionsAreTheDeclarationSplitByModule(t *testing.T) {
in := aSmallMesh(t)
d, _, err := Compose(in, "laptop")
if err != nil {
t.Fatal(err)
}
var want []string
for _, r := range d.Resources {
want = append(want, r["type"].(string)+":"+r["id"].(string))
}
sort.Strings(want)
fps := map[string]string{"mesh-wireguard": fpFor('a'), "files": fpFor('b')}
out, skipped, err := Compositions(in, "laptop", fps, Foreseeing)
if err != nil || len(skipped) != 0 {
t.Fatalf("Compositions = %v, skipped %v", err, skipped)
}
var got []string
for key, c := range out {
node, module, fp, isNode, err := delivery.ParseComposedKey(key)
if err != nil || node != "laptop" || module != c.Module || fp != c.FP || (isNode != (module == "")) {
t.Errorf("key %s holds %s at %s (%v)", key, c.Module, c.FP, err)
}
if !isNode && fp != fps[module] {
t.Errorf("%s is composed at %s, given %s", module, fp, fps[module])
}
if _, err := delivery.EncodeComposed(c); err != nil {
t.Errorf("%s: %v", key, err)
}
for _, raw := range c.Resources {
var r struct{ ID, Type, Module string }
_ = json.Unmarshal(raw, &r)
if r.Module != module {
t.Errorf("%s holds %s of %q", key, r.ID, r.Module)
}
got = append(got, r.Type+":"+r.ID)
}
}
sort.Strings(got)
if strings.Join(got, " ") != strings.Join(want, " ") {
t.Errorf("the compositions hold %v, the declaration %v", got, want)
}
one, skipped, err := Compositions(in, "laptop", map[string]string{"files": fpFor('b')}, Foreseeing)
if err != nil || strings.Join(skipped, ",") != "mesh-wireguard" {
t.Fatalf("one module: %v, skipped %v", err, skipped)
}
for _, c := range one {
if c.Module == "mesh-wireguard" {
t.Errorf("a module given no fingerprint was composed")
}
}
if _, _, err := Compositions(in, "nowhere", fps, Foreseeing); err == nil {
t.Errorf("a node the mesh does not have was composed")
}
}
// 1943/AC1: the upgrade policies and sources a composition's value may name are the catalogue's own words.
func TestTheComposedPoliciesAreTheCatalogues(t *testing.T) {
for _, p := range []string{catalogue.PolicyRoll, catalogue.PolicyTogether, catalogue.PolicyRecord} {
for _, from := range []string{catalogue.FromPerson, catalogue.FromModule, catalogue.FromBus, catalogue.FromData, catalogue.FromDefault} {
v := delivery.ComposedDigest{Digest: "sha256:" + strings.Repeat("a", 64), Judgement: delivery.JudgedNeither, Policy: p, PolicyFrom: from}
if _, err := delivery.EncodeComposedDigest(v); err != nil {
t.Errorf("%s from %s: %v", p, from, err)
}
}
}
if delivery.PolicyRecord != catalogue.PolicyRecord {
t.Errorf("delivery.PolicyRecord = %q, the catalogue's %q", delivery.PolicyRecord, catalogue.PolicyRecord)
}
}
+47
View File
@@ -0,0 +1,47 @@
package declaration
import (
"encoding/json"
"fmt"
)
// ParseResource reads one resource as Parse reads each of a declaration's resources: by its type, strictly, refusing
// a field its kind does not use and anything its kind refuses, and never an action (the link may not carry one, ADR
// 0005). It judges the resource alone: what Parse judges across resources (an opening on a node not adopted, the
// files the provisioner fills, what replaces what) is for the whole a node applies (novox/hq ADR 0539 clause 1: a
// composition is one module's resources, and a node applies its compositions together with its own key).
func ParseResource(raw json.RawMessage) (Resource, error) {
var head struct {
ID string `json:"id"`
Type Type `json:"type"`
}
if err := json.Unmarshal(raw, &head); err != nil {
return nil, &RefusalError{Problems: []string{"not a resource: " + err.Error()}}
}
where := "resource"
if head.ID == "" {
return nil, &RefusalError{Problems: []string{where + ": no id. Identity is what lets the host know this is " +
"the same resource it applied last time"}}
}
where = fmt.Sprintf("resource %q", head.ID)
resource := newOf(head.Type)
if resource == nil {
return nil, &RefusalError{Problems: []string{fmt.Sprintf("%s: unknown type %q. This host understands %s",
where, head.Type, vocabulary())}}
}
if unknown := unknownFields(raw, resource); len(unknown) > 0 {
var problems []string
for _, field := range unknown {
problems = append(problems, fmt.Sprintf("%s: a %s does not use %q, and it is set. Refused rather "+
"than ignored", where, head.Type, field))
}
return nil, &RefusalError{Problems: problems}
}
if err := json.Unmarshal(raw, resource); err != nil {
return nil, &RefusalError{Problems: []string{fmt.Sprintf("%s: %s", where, err)}}
}
if problems := resource.validate(where, false); len(problems) > 0 {
return nil, &RefusalError{Problems: problems}
}
return resource, nil
}
+118 -16
View File
@@ -1,6 +1,7 @@
package delivery
import (
"bytes"
"encoding/json"
"fmt"
"reflect"
@@ -16,6 +17,10 @@ import (
// only compared; the node agent applying them is release 6 (ADR 0429 §7). Only the assigner's credential writes these
// buckets; the watchman, the view and the controller read them.
//
// **Release 6** (ADR 0539 clauses 1, 6 and 16): a want also names the digest of the one composition it allows, written
// only once that composition exists (the seventh rule, RuleComposed), and carries the answer to a failed backup; the
// node's own want `want.<node>.node` names the digest of the node's own composition (NodeWant).
//
// **The bucket is the first word, the key the rest** (as the builder's `builds.<module>.<fp>` is the key
// `<module>.<fp>` in its builds bucket): the differences bucket's keys are `<node>.<module>`. The wants bucket holds
// two kinds of key, so each keeps its word: `want.<node>.<module>` and `held.<module>.<fingerprint hex>`.
@@ -50,7 +55,33 @@ func WantKey(node, module string) (string, error) {
return wantWord + "." + have, nil
}
// ParseWantKey is the inverse of WantKey; it refuses a held key, a have key and any key of another bucket, naming it.
// WantNodeKey is the key, in AssignerWantsBucket, of a node's own want: `want.<node>.node`, naming the digest of the
// node's own composition (ADR 0539 clause 6; ADR 0429 §4's node key). The assigner writes it once the node's own
// composition exists.
func WantNodeKey(node string) (string, error) {
own, err := HaveNodeKey(node)
if err != nil {
return "", err
}
return wantWord + "." + own, nil
}
// ParseWantNodeKey is the inverse of WantNodeKey; it refuses a module's want key and any key of another bucket,
// naming it.
func ParseWantNodeKey(key string) (node string, err error) {
rest, ok := strings.CutPrefix(key, wantWord+".")
if !ok {
return "", fmt.Errorf("delivery: %q is not a node's own want key, want.<node>.node", key)
}
node, _, isNode, err := ParseHaveKey(rest)
if err != nil || !isNode {
return "", fmt.Errorf("delivery: %q is not a node's own want key, want.<node>.node", key)
}
return node, nil
}
// ParseWantKey is the inverse of WantKey; it refuses a held key, a have key, the node's own want key and any key of
// another bucket, naming it.
func ParseWantKey(key string) (node, module string, err error) {
rest, ok := strings.CutPrefix(key, wantWord+".")
if !ok {
@@ -100,7 +131,8 @@ func ParseDifferenceKey(key string) (node, module string, err error) {
return node, module, nil
}
// The six rules of ADR 0429 §6, by number, as a want or a difference names the one that holds a node back.
// The rules of ADR 0429 §6 and ADR 0539 clause 6, by number, as a want or a difference names the one that holds a node
// back.
const (
RuleBuildDone = 1 // the build of F is done
RuleNotHeld = 2 // F is not held
@@ -108,15 +140,25 @@ const (
RuleStep = 4 // at most step other nodes are ahead (step 0: no limit, ADR 0528 §3)
RuleProviders = 5 // every provider the module requires runs healthy on that node, or at its canary
RuleGrant = 6 // the module's credential for F exists (grants.<module>.<fp>)
RuleComposed = 7 // the module's composition for that node at F exists (mesh-controller_composed, ADR 0539 §6)
)
func checkRule(rule int, of string) error {
if rule < RuleBuildDone || rule > RuleGrant {
return fmt.Errorf("delivery: %s names the rule %d, which is not one of the six (1 to 6)", of, rule)
if rule < RuleBuildDone || rule > RuleComposed {
return fmt.Errorf("delivery: %s names the rule %d, which is not one of the seven (1 to 7)", of, rule)
}
return nil
}
// The answers a want carries to a failed backup before an apply (ADR 0539 clause 16): the node agent reads them from
// its own want and from nothing else.
const (
// BackupAgain is the answer to back the module up again before the apply.
BackupAgain = "again"
// BackupSkip is the answer to go on without a backup.
BackupSkip = "skip"
)
func checkFingerprint(fp, of string) error {
if _, err := FingerprintKeyPart(fp); err != nil {
return fmt.Errorf("delivery: %s: %w", of, err)
@@ -124,29 +166,54 @@ func checkFingerprint(fp, of string) error {
return nil
}
// Want is a node's want for a module: the fingerprint FP the node should run, or Removed for a module that left the
// node's modules (then FP is empty). When the module's newest fingerprint is not yet allowed on this node, Newest names
// it and Rule the rule that holds the node back; both are empty when FP is the newest. Words say why in a sentence.
// The entry's own time says since when.
// Want is a node's want for a module: the fingerprint FP the node should run and the digest Composed of the one
// composition of it the node may apply (ADR 0539 clause 1), or Removed for a module that left the node's modules (then
// both are empty). When the module's newest fingerprint is not yet allowed on this node, Newest names it and Rule the
// rule that holds the node back; both are empty when FP is the newest. Backup is the answer to a failed backup before
// the apply, BackupAgain or BackupSkip (clause 16), empty when none was asked. Words say why in a sentence. The
// entry's own time says since when.
type Want struct {
V int `json:"v"`
FP string `json:"fp,omitempty"`
Removed bool `json:"removed,omitempty"`
Newest string `json:"newest,omitempty"`
Rule int `json:"rule,omitempty"`
Words string `json:"words,omitempty"`
V int `json:"v"`
FP string `json:"fp,omitempty"`
Composed string `json:"composed,omitempty"`
Removed bool `json:"removed,omitempty"`
Newest string `json:"newest,omitempty"`
Rule int `json:"rule,omitempty"`
Backup string `json:"backup,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 {
if w.Removed {
if w.FP != "" || w.Newest != "" || w.Rule != 0 {
return fmt.Errorf("delivery: a removed want names no fingerprint and no rule")
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, backup answer or put")
}
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 {
return err
}
if w.Composed == "" {
return fmt.Errorf("delivery: the want of %s names no composition: a want names the digest of the one composition it allows (ADR 0539 clause 1)", w.FP)
}
if err := checkDigest(w.Composed, "a want's composition digest"); err != nil {
return err
}
switch w.Backup {
case "", BackupAgain, BackupSkip:
default:
return fmt.Errorf("delivery: the backup answer %q is not %s or %s", w.Backup, BackupAgain, BackupSkip)
}
if w.Newest == "" && w.Rule == 0 {
return nil
}
@@ -236,6 +303,41 @@ func DecodeWant(data []byte) (Want, error) {
// SameWant says whether two wants are the same value, so the assigner writes only when they differ.
func SameWant(a, b Want) bool { return a == b }
// NodeWant is a node's own want (WantNodeKey): the digest Composed of the node's own composition it may apply. Words
// say why in a sentence.
type NodeWant struct {
V int `json:"v"`
Composed string `json:"composed"`
Words string `json:"words,omitempty"`
}
// EncodeNodeWant is w as JSON at Version; it refuses a digest not of the sha256 form.
func EncodeNodeWant(w NodeWant) ([]byte, error) {
if err := checkDigest(w.Composed, "the node's own want's composition digest"); err != nil {
return nil, err
}
w.V = Version
return json.Marshal(w)
}
// DecodeNodeWant reads a node's own want; it refuses an unknown version (ErrUnknownVersion), a field a node's own want
// does not have (a fingerprint among them), and what EncodeNodeWant refuses.
func DecodeNodeWant(data []byte) (NodeWant, error) {
if err := checkVersion(data); err != nil {
return NodeWant{}, err
}
var w NodeWant
d := json.NewDecoder(bytes.NewReader(data))
d.DisallowUnknownFields()
if err := d.Decode(&w); err != nil {
return NodeWant{}, fmt.Errorf("delivery: a node's own want does not fit its shape: %w", err)
}
if err := checkDigest(w.Composed, "the node's own want's composition digest"); err != nil {
return NodeWant{}, err
}
return w, nil
}
// EncodeHeld is h as JSON at Version; it refuses a node name a key refuses.
func EncodeHeld(h Held) ([]byte, error) {
if err := h.check(); err != nil {
+55 -3
View File
@@ -4,6 +4,7 @@ import (
"errors"
"strings"
"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,
@@ -74,9 +75,10 @@ func TestDifferenceKey(t *testing.T) {
// 1297/B1a: wants, holds and differences round-trip through their encoders unchanged, at Version.
func TestAssignerValuesRoundTrip(t *testing.T) {
wants := []Want{
{FP: fp2of('a')},
{FP: fp2of('a'), Newest: fp2of('b'), Rule: RuleCanary, Words: "the canary ace does not run it healthy yet"},
{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"},
{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 {
data, err := EncodeWant(w)
@@ -89,6 +91,24 @@ func TestAssignerValuesRoundTrip(t *testing.T) {
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"}
data, err := EncodeHeld(h)
if err != nil {
@@ -127,7 +147,10 @@ func TestAssignerValuesRefused(t *testing.T) {
{Removed: true, FP: fp2of('a')},
{FP: fp2of('a'), Newest: fp2of('b')},
{FP: fp2of('a'), Newest: fp2of('a'), Rule: 3},
{FP: fp2of('a'), Newest: fp2of('b'), Rule: 7},
{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 {
t.Errorf("EncodeWant(%+v) accepted", w)
@@ -148,6 +171,19 @@ func TestAssignerValuesRefused(t *testing.T) {
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) {
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 {
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)
}
}
}
+241
View File
@@ -0,0 +1,241 @@
package delivery
import (
"encoding/json"
"fmt"
"regexp"
"strings"
)
// The controller's compositions (novox/hq ADR 0539 clauses 1 to 4). A **composition** is one module's resources
// composed for one node at one fingerprint, written by the controller, which keeps the inputs. It lives in a bucket of
// each node, `composed-<node>`, which only that node's node-engine reads (a composition can carry a sealed value), under
// `<node>.<module>.<fingerprint hex>`, and the node's own resources (adoption, firewall openings, accounts, the
// resolver, identities and groups) under `<node>.node`. Beside it, under the same key in `mesh-controller_composed`,
// which the assigner reads, the controller writes the composition's digest and its judgement of the move from the
// composition the node runs now (ComposedDigest). The controller writes both by compare-and-set while it holds the
// lease; the operator's credential is a second writer on the hand path alone (clause 10, mesh-hand-compose).
//
// A composition is derived state (clause 4): its key and its digest are deleted by the controller when its fingerprint
// is no longer among the module's five newest and no want names it, and each bucket keeps five revisions of a key.
const (
// ComposedBucketPrefix starts the name of a node's bucket of compositions, ComposedBucketFor(node).
ComposedBucketPrefix = "composed-"
// ComposedHistory is the revisions a node's composition bucket keeps per key (ADR 0539 clause 4).
ComposedHistory = 5
// ControllerModule is the module of the controller, the writer of the compositions.
ControllerModule = "mesh-controller"
// ControllerComposedBucket is the controller's bucket of each composition's digest and judgement, `<owner
// module>_<name>`, read by the assigner (ADR 0539 clauses 2 and 6).
ControllerComposedBucket = ControllerModule + "_composed"
// ControllerComposedHistory is the revisions that bucket keeps per key.
ControllerComposedHistory = 5
// ComposedKeep is how many of a module's newest fingerprints keep their compositions though no want names them
// (ADR 0539 clause 4).
ComposedKeep = 5
)
// nodeWord is the last part of a node's own key, in the have bucket and in a node's composition bucket alike.
const nodeWord = "node"
var bucketNode = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
// ComposedBucketFor is the name of a node's bucket of compositions, "composed-<node>", as declaration.FileBucketFor
// names a node's object bucket (ADR 0488 §4). The node name must be one a bucket and a subject token accept: letters,
// digits, '_' and '-'.
func ComposedBucketFor(node string) (string, error) {
if node == "" {
return "", fmt.Errorf("delivery: the node name is empty; a node's composition bucket is named for its node")
}
if !bucketNode.MatchString(node) {
return "", fmt.Errorf("delivery: the node name %q is not one a bucket accepts: letters, digits, '_' and '-' only", node)
}
return ComposedBucketPrefix + node, nil
}
// ComposedKey is the key, in ComposedBucketFor(node) and in ControllerComposedBucket, of a module's composition for a
// node at a fingerprint: `<node>.<module>.<hex>`, the hex being the fingerprint's key part (FingerprintKeyPart). A
// module named `node` is refused, since `<node>.node` is the node's own key.
func ComposedKey(node, module, fingerprint string) (string, error) {
have, err := HaveKey(node, module)
if err != nil {
return "", err
}
part, err := FingerprintKeyPart(fingerprint)
if err != nil {
return "", err
}
return have + "." + part, nil
}
// ComposedNodeKey is the key of a node's own composition: `<node>.node`.
func ComposedNodeKey(node string) (string, error) {
return HaveNodeKey(node)
}
// ParseComposedKey is the inverse of ComposedKey and ComposedNodeKey: isNode is true for `<node>.node`, with module
// and fingerprint empty. It refuses any other shape, naming the key.
func ParseComposedKey(key string) (node, module, fingerprint string, isNode bool, err error) {
refused := func(why error) (string, string, string, bool, error) {
return "", "", "", false, fmt.Errorf("delivery: %q is not a composition key, <node>.<module>.<fingerprint hex> or <node>.node: %w", key, why)
}
parts := strings.Split(key, ".")
switch len(parts) {
case 2:
if parts[1] != nodeWord {
return refused(fmt.Errorf("a module's composition names its fingerprint"))
}
if err := checkName("node", parts[0]); err != nil {
return refused(err)
}
return parts[0], "", "", true, nil
case 3:
node, module, isNode, err := ParseHaveKey(parts[0] + "." + parts[1])
if err != nil {
return refused(err)
}
if isNode {
return refused(fmt.Errorf("the node's own key names no fingerprint"))
}
fp, err := ParseFingerprintKeyPart(parts[2])
if err != nil {
return refused(err)
}
return node, module, fp, false, nil
}
return refused(fmt.Errorf("it has %d parts", len(parts)))
}
// Judgement is the controller's judgement of a move from the composition a node runs, as its have names it, to a new
// one (ADR 0539 clauses 2 and 14, by ADR 0377 §3's tests).
type Judgement string
// The judgements of a move.
const (
// JudgedRecreates is a move that recreates a container.
JudgedRecreates Judgement = "recreates"
// JudgedMovesData is a move that gives the module's data another place.
JudgedMovesData Judgement = "moves-data"
// JudgedNeither is a move that does neither, and the judgement of a composition the node never ran.
JudgedNeither Judgement = "neither"
// JudgedCannot is a move that cannot be judged, because the composition the node runs is gone or cannot be read.
JudgedCannot Judgement = "cannot-judge"
// JudgedUnchanged is a composition equal to the module's slice of the declaration the node last applied, for a node
// whose have names no composition yet (the first want at the switch, novox/hq ADR 0540 clause 9): applying it
// changes nothing on the node, so it needs no person's yes.
JudgedUnchanged Judgement = "unchanged"
)
var digestForm = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
func checkDigest(digest, of string) error {
if !digestForm.MatchString(digest) {
return fmt.Errorf("delivery: %s %q is not sha256:<64 lowercase hex digits>", of, digest)
}
return nil
}
// ComposedDigest is the value, in ControllerComposedBucket, beside a composition: its Digest (ComposedDigestOf), and
// the Judgement of the move to it from the composition the node runs now, whose digest is Against (empty when the node
// runs no composition of the module). Words say why, and are required for a move that cannot be judged. Rewritten when
// the node's have changes.
//
// Policy is the module's upgrade policy as the controller holds it (ADR 0236, ADR 0539 clause 14), with PolicyFrom
// where it comes from and PolicyWhy why: the assigner holds a move of a module of policy record from it, the operator's
// own choice (`upgrade` with policy) included. Rewritten when the policy changes. Empty on a composition made by hand
// (mesh-hand-compose), whose want the operator writes.
type ComposedDigest struct {
V int `json:"v"`
Digest string `json:"digest"`
Judgement Judgement `json:"judgement"`
Against string `json:"against,omitempty"`
Words string `json:"words,omitempty"`
// UserListOnly says the composition differs from the one the node's have names only in the bus's user list (the
// bus module's file `<module>.bus-users`), whatever the judgement says of its mount: the assigner writes such a
// revision into the bus node's want at once, outside the record hold, at the fingerprint the node runs (ADR 0539
// clause 7).
UserListOnly bool `json:"user-list-only,omitempty"`
Policy string `json:"policy,omitempty"`
PolicyFrom string `json:"policy-from,omitempty"`
PolicyWhy string `json:"policy-why,omitempty"`
}
// The upgrade policies and their sources a composition's value names: catalogue.PolicyRoll, PolicyTogether and
// PolicyRecord, and catalogue.FromPerson, FromModule, FromBus, FromData and FromDefault, the same words (the compose
// package's test holds them equal; this package does not import the catalogue).
var (
composedPolicies = []string{"roll", "together", "record"}
composedPolicyFroms = []string{"person", "module", "the bus", "irreplaceable data", "default"}
)
func oneOf(v string, of []string) bool {
for _, o := range of {
if v == o {
return true
}
}
return false
}
func (c ComposedDigest) check() error {
if err := checkDigest(c.Digest, "the composition's digest"); err != nil {
return err
}
if c.Against != "" {
if err := checkDigest(c.Against, "the digest a move is judged against"); err != nil {
return err
}
}
switch c.Judgement {
case JudgedRecreates, JudgedMovesData:
if c.Against == "" {
return fmt.Errorf("delivery: a move judged %s names the composition it is judged against", c.Judgement)
}
case JudgedNeither, JudgedUnchanged:
case JudgedCannot:
if c.Words == "" {
return fmt.Errorf("delivery: a move that cannot be judged says why")
}
default:
return fmt.Errorf("delivery: the judgement %q is not %s, %s, %s, %s or %s", c.Judgement, JudgedRecreates,
JudgedMovesData, JudgedNeither, JudgedCannot, JudgedUnchanged)
}
if c.Policy != "" || c.PolicyFrom != "" {
if !oneOf(c.Policy, composedPolicies) {
return fmt.Errorf("delivery: the upgrade policy %q is not one of %v", c.Policy, composedPolicies)
}
if !oneOf(c.PolicyFrom, composedPolicyFroms) {
return fmt.Errorf("delivery: the upgrade policy's source %q is not one of %v", c.PolicyFrom, composedPolicyFroms)
}
}
return nil
}
// PolicyRecord is the policy whose moves the assigner holds for a person's yes (ADR 0539 clause 14).
const PolicyRecord = "record"
// EncodeComposedDigest is c as JSON at Version; it refuses an unknown judgement, a digest not of the sha256 form, a
// move judged to recreate or move data against nothing, and a move that cannot be judged without its reason.
func EncodeComposedDigest(c ComposedDigest) ([]byte, error) {
if err := c.check(); err != nil {
return nil, err
}
c.V = Version
return json.Marshal(c)
}
// DecodeComposedDigest reads the value beside a composition; it refuses an unknown version (ErrUnknownVersion) and
// what EncodeComposedDigest refuses.
func DecodeComposedDigest(data []byte) (ComposedDigest, error) {
var c ComposedDigest
if err := decode(data, &c); err != nil {
return ComposedDigest{}, err
}
if err := c.check(); err != nil {
return ComposedDigest{}, err
}
return c, nil
}
// SameComposedDigest says whether two values are the same, so the controller writes only when they differ.
func SameComposedDigest(a, b ComposedDigest) bool { return a == b }
+125
View File
@@ -0,0 +1,125 @@
package delivery
import (
"errors"
"strings"
"testing"
)
// 1943/AC1: a node's composition bucket is named for the node, and a node with no name, or one a bucket refuses,
// is refused, naming why.
func TestComposedBucketFor(t *testing.T) {
b, err := ComposedBucketFor("ace")
if err != nil || b != "composed-ace" {
t.Fatalf("ComposedBucketFor(ace) = %q, %v", b, err)
}
for _, bad := range []string{"", "a.b", "a*", "a b", "a>"} {
if _, err := ComposedBucketFor(bad); err == nil || !strings.Contains(err.Error(), "node") {
t.Errorf("ComposedBucketFor(%q) = %v, want a refusal naming the node", bad, err)
}
}
}
// 1943/AC1: a module's composition key and the node's own key round-trip through one parser; an empty node, a
// module named node, a bad fingerprint and a key of another shape are refused, naming why.
func TestComposedKeys(t *testing.T) {
fp := fp2of('c')
key, err := ComposedKey("ace", "mesh-mail", fp)
if err != nil || key != "ace.mesh-mail."+strings.Repeat("c", 64) {
t.Fatalf("ComposedKey = %q, %v", key, err)
}
node, module, back, isNode, err := ParseComposedKey(key)
if err != nil || node != "ace" || module != "mesh-mail" || back != fp || isNode {
t.Errorf("ParseComposedKey(%q) = %q, %q, %q, %v, %v", key, node, module, back, isNode, err)
}
own, err := ComposedNodeKey("ace")
if err != nil || own != "ace.node" {
t.Fatalf("ComposedNodeKey = %q, %v", own, err)
}
node, module, back, isNode, err = ParseComposedKey(own)
if err != nil || node != "ace" || module != "" || back != "" || !isNode {
t.Errorf("ParseComposedKey(%q) = %q, %q, %q, %v, %v", own, node, module, back, isNode, err)
}
refusals := []struct {
node, module, fp, why string
}{
{"", "mesh-mail", fp, "node name is empty"},
{"ace", "node", fp, `"node" is refused`},
{"ace", "", fp, "module name is empty"},
{"ace", "mesh-mail", "src2:abc", "not a fingerprint of the new form"},
{"ace", "mesh-mail", "src1:" + strings.Repeat("c", 64), "not a fingerprint of the new form"},
{"a.b", "mesh-mail", fp, "node name"},
}
for _, r := range refusals {
if _, err := ComposedKey(r.node, r.module, r.fp); err == nil || !strings.Contains(err.Error(), r.why) {
t.Errorf("ComposedKey(%q, %q, %q) = %v, want a refusal saying %q", r.node, r.module, r.fp, err, r.why)
}
}
if _, err := ComposedNodeKey(""); err == nil {
t.Errorf("ComposedNodeKey(\"\") accepted")
}
for _, bad := range []string{"", "ace", "ace.mesh-mail", "ace.node.x", "ace.mesh-mail.abc", ".mesh-mail." + strings.Repeat("c", 64),
"ace.node." + strings.Repeat("c", 64), "ace.mesh-mail." + strings.Repeat("c", 64) + ".x", "want.ace.mesh-mail"} {
if _, _, _, _, err := ParseComposedKey(bad); err == nil || !strings.Contains(err.Error(), "\""+bad+"\"") {
t.Errorf("ParseComposedKey(%q) = %v, want a refusal naming it", bad, err)
}
}
}
// 1943/AC1: the controller's value beside a composition, its digest and the judgement of the move, round-trips; an
// unknown judgement, a digest not of the sha256 form, a move judged against nothing, a cannot-judge without its
// reason, and an upgrade policy or source outside the catalogue's words are refused, naming why; an unknown version is
// refused as one. The module's upgrade policy and its source ride beside the judgement, for the assigner's hold.
func TestComposedDigestValue(t *testing.T) {
digest := "sha256:" + strings.Repeat("d", 64)
against := "sha256:" + strings.Repeat("e", 64)
for _, v := range []ComposedDigest{
{Digest: digest, Judgement: JudgedNeither},
{Digest: digest, Judgement: JudgedNeither, Against: against},
{Digest: digest, Judgement: JudgedRecreates, Against: against, Words: "the container's image changes"},
{Digest: digest, Judgement: JudgedMovesData, Against: against},
{Digest: digest, Judgement: JudgedCannot, Words: "the composition the node runs is gone"},
{Digest: digest, Judgement: JudgedUnchanged, Words: "equal to the declaration the node last applied"},
{Digest: digest, Judgement: JudgedNeither, Policy: "record", PolicyFrom: "person", PolicyWhy: "the operator: its data"},
{Digest: digest, Judgement: JudgedNeither, Policy: "roll", PolicyFrom: "default"},
{Digest: digest, Judgement: JudgedNeither, Against: against, UserListOnly: true, Words: "only the bus's user list changes"},
} {
raw, err := EncodeComposedDigest(v)
if err != nil {
t.Fatalf("EncodeComposedDigest(%+v): %v", v, err)
}
back, err := DecodeComposedDigest(raw)
v.V = Version
if err != nil || !SameComposedDigest(back, v) {
t.Errorf("round trip of %+v = %+v, %v", v, back, err)
}
}
refusals := []struct {
v ComposedDigest
why string
}{
{ComposedDigest{Digest: digest, Judgement: "maybe"}, `judgement "maybe"`},
{ComposedDigest{Digest: digest}, `judgement ""`},
{ComposedDigest{Digest: "sha256:abc", Judgement: JudgedNeither}, "not sha256"},
{ComposedDigest{Judgement: JudgedNeither}, "not sha256"},
{ComposedDigest{Digest: digest, Judgement: JudgedRecreates}, "against"},
{ComposedDigest{Digest: digest, Judgement: JudgedMovesData}, "against"},
{ComposedDigest{Digest: digest, Judgement: JudgedCannot}, "why"},
{ComposedDigest{Digest: digest, Judgement: JudgedNeither, Against: "sha256:zz"}, "not sha256"},
{ComposedDigest{Digest: digest, Judgement: JudgedNeither, Policy: "sometimes", PolicyFrom: "person"}, `policy "sometimes"`},
{ComposedDigest{Digest: digest, Judgement: JudgedNeither, Policy: "record"}, `source ""`},
{ComposedDigest{Digest: digest, Judgement: JudgedNeither, PolicyFrom: "person"}, `policy ""`},
}
for _, r := range refusals {
if _, err := EncodeComposedDigest(r.v); err == nil || !strings.Contains(err.Error(), r.why) {
t.Errorf("EncodeComposedDigest(%+v) = %v, want a refusal saying %q", r.v, err, r.why)
}
}
if _, err := DecodeComposedDigest([]byte(`{"v":2,"digest":"` + digest + `","judgement":"neither"}`)); !errors.Is(err, ErrUnknownVersion) {
t.Errorf("version 2 = %v, want ErrUnknownVersion", err)
}
if _, err := DecodeComposedDigest([]byte(`{"v":1,"digest":"` + digest + `","judgement":"perhaps"}`)); err == nil {
t.Errorf("an unknown judgement was read")
}
}
+100
View File
@@ -0,0 +1,100 @@
package delivery
import (
"crypto/ed25519"
"encoding/base64"
"fmt"
)
// A composition's signature (novox/hq ADR 0539 clause 3): the controller signs each composition it writes with its
// key, and the operator signs one made by hand with mesh-hand-compose with theirs. The node-engine trusts exactly these two
// signers and applies nothing of a composition whose signature fails or whose signer is neither. The signature is an
// ed25519 signature, as a declaration's is, of the composition's digest (ComposedDigestOf) after a context line, so a
// signature made for anything else never verifies here.
const composedSigningContext = "novox composition v1\n"
// Signature is who signed a composition and the signature: Signer the signer's ed25519 public key and Sig the
// signature, both in standard base64.
type Signature struct {
Signer string `json:"signer"`
Sig string `json:"sig"`
}
func (s Signature) check() error {
if key, err := base64.StdEncoding.DecodeString(s.Signer); err != nil || len(key) != ed25519.PublicKeySize {
return fmt.Errorf("delivery: a composition's signer %q is not an ed25519 public key in base64", s.Signer)
}
if sig, err := base64.StdEncoding.DecodeString(s.Sig); err != nil || len(sig) != ed25519.SignatureSize {
return fmt.Errorf("delivery: a composition's signature is not an ed25519 signature in base64")
}
return nil
}
// SignComposed is c signed with key: its digest signed, and the key's public half named as the signer. A signature c
// already carries is replaced.
func SignComposed(c Composed, key ed25519.PrivateKey) (Composed, error) {
if len(key) != ed25519.PrivateKeySize {
return Composed{}, fmt.Errorf("delivery: the key to sign a composition with is not an ed25519 private key")
}
return SignComposedBy(c, key.Public().(ed25519.PublicKey), func(message []byte) ([]byte, error) {
return ed25519.Sign(key, message), nil
})
}
// SignComposedBy is SignComposed for a signer that keeps its private half to itself (the controller's identity store
// signs a message and hands back the signature): sign signs the message it is given with the key whose public half is
// signer. A signature that does not verify against signer is refused, so a composition is never written signed by a
// key other than the one it names.
func SignComposedBy(c Composed, signer ed25519.PublicKey, sign func(message []byte) ([]byte, error)) (Composed, error) {
if len(signer) != ed25519.PublicKeySize {
return Composed{}, fmt.Errorf("delivery: the signer of a composition is not an ed25519 public key")
}
c.Signature = nil
digest, err := ComposedDigestOf(c)
if err != nil {
return Composed{}, err
}
message := []byte(composedSigningContext + digest)
sig, err := sign(message)
if err != nil {
return Composed{}, err
}
if !ed25519.Verify(signer, message, sig) {
return Composed{}, fmt.Errorf("delivery: the signature made for the composition of %q on %s does not verify against its signer", c.Module, c.Node)
}
c.Signature = &Signature{Signer: base64.StdEncoding.EncodeToString(signer), Sig: base64.StdEncoding.EncodeToString(sig)}
return c, nil
}
// VerifyComposed is c's digest when c is signed by one of the trusted keys and its signature verifies against its
// digest; otherwise it refuses, saying why, and nothing of c is to be applied: an unsigned composition, one whose
// signer is not a trusted key, and one changed after it was signed.
func VerifyComposed(c Composed, trusted []ed25519.PublicKey) (string, error) {
if c.Signature == nil {
return "", fmt.Errorf("delivery: the composition of %q on %s is unsigned, and only a signed one is applied", c.Module, c.Node)
}
if err := c.Signature.check(); err != nil {
return "", err
}
signer, _ := base64.StdEncoding.DecodeString(c.Signature.Signer)
var key ed25519.PublicKey
for _, t := range trusted {
if t.Equal(ed25519.PublicKey(signer)) {
key = t
}
}
if key == nil {
return "", fmt.Errorf("delivery: the composition of %q on %s is signed by %s, which is not a trusted signer (the controller's key or the operator's)",
c.Module, c.Node, c.Signature.Signer)
}
digest, err := ComposedDigestOf(c)
if err != nil {
return "", err
}
sig, _ := base64.StdEncoding.DecodeString(c.Signature.Sig)
if !ed25519.Verify(key, []byte(composedSigningContext+digest), sig) {
return "", fmt.Errorf("delivery: the composition of %q on %s does not match its signature: it was changed after it was signed, or signed for another",
c.Module, c.Node)
}
return digest, nil
}
+103
View File
@@ -0,0 +1,103 @@
package delivery
import (
"crypto/ed25519"
"crypto/rand"
"encoding/json"
"strings"
"testing"
)
func aSigningKey(t *testing.T) (ed25519.PublicKey, ed25519.PrivateKey) {
t.Helper()
pub, priv, err := ed25519.GenerateKey(rand.Reader)
if err != nil {
t.Fatal(err)
}
return pub, priv
}
func aComposition() Composed {
return Composed{Node: "ace", Module: "m", FP: fp2of('a'), Resources: []json.RawMessage{
json.RawMessage(`{"id":"m.f","type":"file","module":"m","path":"/etc/m","content":"x"}`)}}
}
// 1943/AC5: a composition signed by the controller's key or by the operator's verifies against the two trusted keys,
// keeps its digest, and survives its encoding; one signed by another key, unsigned, or changed by one byte after it was
// signed does not verify, and nothing of it is to be applied.
func TestACompositionIsSignedByATrustedKey(t *testing.T) {
controllerPub, controller := aSigningKey(t)
operatorPub, operator := aSigningKey(t)
_, stranger := aSigningKey(t)
trusted := []ed25519.PublicKey{controllerPub, operatorPub}
c := aComposition()
want, _ := ComposedDigestOf(c)
for name, key := range map[string]ed25519.PrivateKey{"controller": controller, "operator": operator} {
signed, err := SignComposed(c, key)
if err != nil {
t.Fatalf("%s: %v", name, err)
}
raw, err := EncodeComposed(signed)
if err != nil {
t.Fatal(err)
}
back, err := DecodeComposed(raw)
if err != nil {
t.Fatal(err)
}
digest, err := VerifyComposed(back, trusted)
if err != nil || digest != want {
t.Errorf("%s's signature: %q, %v; want the digest %s", name, digest, err, want)
}
if d, _ := ComposedDigestOf(back); d != want {
t.Errorf("%s's signature changed the digest: %s", name, d)
}
changed := strings.Replace(string(raw), `"content":"x"`, `"content":"y"`, 1)
if changed == string(raw) {
t.Fatal("the test changed nothing")
}
if tampered, err := DecodeComposed([]byte(changed)); err != nil {
t.Fatal(err)
} else if _, err := VerifyComposed(tampered, trusted); err == nil || !strings.Contains(err.Error(), "signature") {
t.Errorf("%s: a composition changed by one byte verified: %v", name, err)
}
}
byStranger, _ := SignComposed(c, stranger)
if _, err := VerifyComposed(byStranger, trusted); err == nil || !strings.Contains(err.Error(), "not a trusted") {
t.Errorf("a stranger's signature = %v, want refused as no trusted signer", err)
}
if _, err := VerifyComposed(c, trusted); err == nil || !strings.Contains(err.Error(), "unsigned") {
t.Errorf("an unsigned composition = %v, want refused", err)
}
signed, _ := SignComposed(c, controller)
forged := signed
forged.Signature = &Signature{Signer: signed.Signature.Signer, Sig: byStranger.Signature.Sig}
if _, err := VerifyComposed(forged, trusted); err == nil {
t.Errorf("a stranger's signature under the controller's name verified")
}
if _, err := EncodeComposed(Composed{Node: "ace", Module: "m", FP: fp2of('a'), Resources: c.Resources,
Signature: &Signature{Signer: "not base64!", Sig: "x"}}); err == nil {
t.Errorf("a signature naming no key was encoded")
}
}
// 1943/AC5: a signer that keeps its private half signs through a function, and the composition verifies as one signed
// with the key; a function that signs with another key than the one named is refused.
func TestACompositionIsSignedByAKeptKey(t *testing.T) {
pub, priv := aSigningKey(t)
_, other := aSigningKey(t)
c := aComposition()
signed, err := SignComposedBy(c, pub, func(m []byte) ([]byte, error) { return ed25519.Sign(priv, m), nil })
if err != nil {
t.Fatal(err)
}
if _, err := VerifyComposed(signed, []ed25519.PublicKey{pub}); err != nil {
t.Errorf("a composition signed through the function does not verify: %v", err)
}
if _, err := SignComposedBy(c, pub, func(m []byte) ([]byte, error) { return ed25519.Sign(other, m), nil }); err == nil {
t.Errorf("a signature by another key than the one named was accepted")
}
}
+330
View File
@@ -0,0 +1,330 @@
package delivery
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"git.novox.be/novox/mesh-sdk/go/declaration"
)
// NodeOwn is what a node's own composition (`<node>.node`) carries beside its resources: what the declaration says of
// the node as a whole (novox/hq ADR 0539 clause 2: adoption, accounts, identities and groups). A module's composition
// carries none of it.
type NodeOwn struct {
// Adoption says the node is adopted and which of its modules are taken (ADR 0100, ADR 0103); nil for a converged node.
Adoption *declaration.Adoption `json:"adoption,omitempty"`
// Identities is the identity of every module assigned to the node (ADR 0328 §4), Groups and Libraries every library
// it is reached by (ADR 0328 §3).
Identities []declaration.Identity `json:"identities,omitempty"`
Groups []declaration.Group `json:"groups,omitempty"`
Libraries []declaration.Library `json:"libraries,omitempty"`
// Provisioner is the account the provisioner's client runs as on the node (ADR 0349 §4).
Provisioner string `json:"provisioner_account,omitempty"`
}
func (o NodeOwn) empty() bool {
return o.Adoption == nil && len(o.Identities) == 0 && len(o.Groups) == 0 && len(o.Libraries) == 0 && o.Provisioner == ""
}
// Composed is a composition (ADR 0539 clause 1): one module's resources composed for one node at one fingerprint, or,
// with Module and FP empty, the node's own resources and parts (Own). Each resource is a declaration resource as the
// node-engine reads it, saying its module; it is kept as JSON and written canonically (its keys sorted, its spacing
// dropped), so its digest (ComposedDigestOf) is a function of its content alone. Data says the module declares data,
// so the node agent backs it up before it moves (ADR 0429 §14). Filled names the files of the module the provisioner
// fills (ADR 0349 §5). A file over 64 KiB travels by digest and size (ByDigest), its content in the node's object
// bucket (ADR 0539 clause 4).
type Composed struct {
V int
Node string
Module string
FP string
Data bool
Resources []json.RawMessage
Filled []string
Own NodeOwn
// Signature is the controller's or, made by hand, the operator's (ADR 0539 clause 3, SignComposed); it signs the
// digest and is never part of it.
Signature *Signature
}
// composedWire is a composition as it is written.
type composedWire struct {
V int `json:"v"`
Node string `json:"node"`
Module string `json:"module,omitempty"`
FP string `json:"fp,omitempty"`
Data bool `json:"data,omitempty"`
Resources []json.RawMessage `json:"resources"`
Filled []string `json:"filled_files,omitempty"`
Own *NodeOwn `json:"own,omitempty"`
Signature *Signature `json:"signature,omitempty"`
}
// canonicalJSON is raw with its object keys sorted and its spacing dropped, its numbers kept as written.
func canonicalJSON(raw json.RawMessage) (json.RawMessage, error) {
var v any
d := json.NewDecoder(bytes.NewReader(raw))
d.UseNumber()
if err := d.Decode(&v); err != nil {
return nil, err
}
if _, err := d.Token(); err != io.EOF {
return nil, fmt.Errorf("more after the value")
}
return json.Marshal(v)
}
func (c Composed) check() error {
if _, err := HaveNodeKey(c.Node); err != nil {
return err
}
if c.Module == "" {
if c.FP != "" {
return fmt.Errorf("delivery: the node's own composition names no fingerprint, and this one names %s", c.FP)
}
// It carries the filled paths of its own files, those of a module with no fingerprint of the new form moved
// into it (issue 1945), and never a module's data.
if c.Data {
return fmt.Errorf("delivery: the node's own composition holds no module's data")
}
} else {
if _, err := ComposedKey(c.Node, c.Module, c.FP); err != nil {
return err
}
if !c.Own.empty() {
return fmt.Errorf("delivery: the composition of %s carries the node's own parts, which only <node>.node carries", c.Module)
}
}
seen := map[string]bool{}
files := map[string][]*declaration.File{}
for _, raw := range c.Resources {
r, err := declaration.ParseResource(raw)
if err != nil {
return fmt.Errorf("delivery: the composition of %q on %s: %w", c.Module, c.Node, err)
}
if seen[r.Identity()] {
return fmt.Errorf("delivery: the composition of %q on %s holds the resource %q twice", c.Module, c.Node, r.Identity())
}
seen[r.Identity()] = true
if r.DeclaredModule() != c.Module {
return fmt.Errorf("delivery: the resource %q is of %q, and this is the composition of %q on %s",
r.Identity(), r.DeclaredModule(), c.Module, c.Node)
}
if f, ok := r.(*declaration.File); ok {
if f.TravelsByDigest() {
return fmt.Errorf("delivery: the file %q is over %d bytes and travels by digest (ByDigest), not whole",
f.ID, declaration.FileReferenceThreshold)
}
files[f.Path] = append(files[f.Path], f)
}
}
// A file the provisioner fills says no content, so a composition that does not say it is filled would have the
// node-engine write it empty (issue 1945, live on 2026-10-25): refused, naming it.
filledHere := make(map[string]bool, len(c.Filled))
for _, path := range c.Filled {
filledHere[path] = true
}
for path, fs := range files {
for _, f := range fs {
if f.OwnerAtRuntime == declaration.RuntimeOwned && !filledHere[path] && !saysContent(f) {
return fmt.Errorf("delivery: the file %q says no content and the composition of %q on %s does not say "+
"the provisioner fills %s: the node-engine would write it empty", f.ID, c.Module, c.Node, path)
}
}
}
for _, path := range c.Filled {
if len(files[path]) == 0 {
return fmt.Errorf("delivery: the composition of %q fills %q, and declares no file there", c.Module, path)
}
for _, f := range files[path] {
if f.OwnerAtRuntime == "" {
return fmt.Errorf("delivery: the file %q is filled by the provisioner and says no owner-at-runtime", f.ID)
}
}
}
if c.Signature != nil {
if err := c.Signature.check(); err != nil {
return err
}
}
return nil
}
// saysContent says a file names what to write: content, bytes, a digest or a sealed value. A file the provisioner fills
// names none of them.
func saysContent(f *declaration.File) bool {
return f.Content != "" || f.Bytes != "" || f.Digest != "" || f.Sealed != "" || f.VaultSealed != nil
}
// wire is c checked and made canonical, ready to marshal.
func (c Composed) wire() (composedWire, error) {
if err := c.check(); err != nil {
return composedWire{}, err
}
w := composedWire{V: Version, Node: c.Node, Module: c.Module, FP: c.FP, Data: c.Data, Filled: c.Filled,
Signature: c.Signature, Resources: make([]json.RawMessage, len(c.Resources))}
for i, r := range c.Resources {
canon, err := canonicalJSON(r)
if err != nil {
return composedWire{}, fmt.Errorf("delivery: a resource of %q is not JSON: %w", c.Module, err)
}
w.Resources[i] = canon
}
if !c.Own.empty() {
own := c.Own
w.Own = &own
}
return w, nil
}
// EncodeComposed is c as JSON at Version, each resource canonical. It refuses a composition whose node or fingerprint
// a key refuses, a resource its kind refuses (an action among them), a resource of another module, a module's
// composition carrying the node's own parts, a file over 64 KiB carried whole, and a filled path no file declares.
func EncodeComposed(c Composed) ([]byte, error) {
w, err := c.wire()
if err != nil {
return nil, err
}
return json.Marshal(w)
}
// DecodeComposed reads a composition; it refuses an unknown version (ErrUnknownVersion), a field it does not know, and
// what EncodeComposed refuses.
func DecodeComposed(data []byte) (Composed, error) {
if err := checkVersion(data); err != nil {
return Composed{}, err
}
var w composedWire
d := json.NewDecoder(bytes.NewReader(data))
d.DisallowUnknownFields()
if err := d.Decode(&w); err != nil {
return Composed{}, fmt.Errorf("delivery: a composition does not fit its shape: %w", err)
}
c := Composed{V: w.V, Node: w.Node, Module: w.Module, FP: w.FP, Data: w.Data, Resources: w.Resources,
Filled: w.Filled, Signature: w.Signature}
if w.Own != nil {
c.Own = *w.Own
}
if err := c.check(); err != nil {
return Composed{}, err
}
return c, nil
}
// ComposedDigestOf is a composition's digest, "sha256:<hex>", of its encoding without its signature. A want names it
// (ADR 0539 clause 1), the controller writes it beside the composition, and the signature signs it.
func ComposedDigestOf(c Composed) (string, error) {
c.Signature = nil
raw, err := EncodeComposed(c)
if err != nil {
return "", err
}
sum := sha256.Sum256(raw)
return "sha256:" + hex.EncodeToString(sum[:]), nil
}
// SameComposed says whether two compositions are the same value, their signatures included, so the controller writes
// only when they differ.
func SameComposed(a, b Composed) bool {
ra, errA := EncodeComposed(a)
rb, errB := EncodeComposed(b)
return errA == nil && errB == nil && bytes.Equal(ra, rb)
}
// ByDigest is c with every file over 64 KiB given by its digest and size, and the content of each by digest, for the
// writer to put in the node's object bucket (declaration.FileBucketFor) under declaration.ObjectName before it writes
// the composition (ADR 0539 clause 4, ADR 0488 §3). A file that is sealed, filled with secrets or written into another
// stays as it is (declaration.File.TravelsByDigest).
func (c Composed) ByDigest() (Composed, map[string][]byte, error) {
out := c
out.Resources = make([]json.RawMessage, len(c.Resources))
objects := map[string][]byte{}
for i, raw := range c.Resources {
out.Resources[i] = raw
r, err := declaration.ParseResource(raw)
if err != nil {
return Composed{}, nil, err
}
f, ok := r.(*declaration.File)
if !ok || !f.TravelsByDigest() {
continue
}
ref, body, err := f.AsReference()
if err != nil {
return Composed{}, nil, err
}
var fields map[string]json.RawMessage
if err := json.Unmarshal(raw, &fields); err != nil {
return Composed{}, nil, err
}
delete(fields, "content")
delete(fields, "bytes")
fields["digest"], _ = json.Marshal(ref.Digest)
fields["size"], _ = json.Marshal(ref.Size)
rewritten, err := json.Marshal(fields)
if err != nil {
return Composed{}, nil, err
}
out.Resources[i] = rewritten
objects[ref.Digest] = body
}
return out, objects, nil
}
// SplitComposed splits a node's composed resources into its compositions, by key (ComposedKey, ComposedNodeKey): each
// resource into the composition of the module it says, under that module's fingerprint in fps, and a resource that
// says no module into the node's own, with own. Each filled path goes with the module whose file declares it. The order
// of the resources is kept within each composition. A resource of a module fps names no fingerprint for is refused,
// naming it. Data is the caller's to set, from the module's manifest.
func SplitComposed(node string, resources []json.RawMessage, fps map[string]string, filled []string, own NodeOwn) (map[string]Composed, error) {
nodeKey, err := ComposedNodeKey(node)
if err != nil {
return nil, err
}
out := map[string]Composed{nodeKey: {Node: node, Own: own, Resources: []json.RawMessage{}}}
pathOf := map[string]string{}
for _, raw := range resources {
var head struct {
ID string `json:"id"`
Type string `json:"type"`
Module string `json:"module"`
Path string `json:"path"`
}
if err := json.Unmarshal(raw, &head); err != nil {
return nil, fmt.Errorf("delivery: a resource of %s is not JSON: %w", node, err)
}
key := nodeKey
if head.Module != "" {
fp, ok := fps[head.Module]
if !ok {
return nil, fmt.Errorf("delivery: the resource %q is of %q, whose fingerprint on %s is not given", head.ID, head.Module, node)
}
if key, err = ComposedKey(node, head.Module, fp); err != nil {
return nil, err
}
if _, ok := out[key]; !ok {
out[key] = Composed{Node: node, Module: head.Module, FP: fp}
}
}
if head.Type == string(declaration.TypeFile) && head.Module != "" {
pathOf[head.Path] = key
}
c := out[key]
c.Resources = append(c.Resources, raw)
out[key] = c
}
for _, path := range filled {
key, ok := pathOf[path]
if !ok {
return nil, fmt.Errorf("delivery: the provisioner fills %q on %s, and no module's file declares it", path, node)
}
c := out[key]
c.Filled = append(c.Filled, path)
out[key] = c
}
return out, nil
}
+364
View File
@@ -0,0 +1,364 @@
package delivery
import (
"bytes"
"compress/gzip"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"sort"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/declaration"
)
// capturedDeclaration is a node's declaration as the controller's `plan` verb answered it on 2026-10-24 at 16:19
// (testdata/declaration-<node>.json.gz; the resources carrying a sealed value or a secret taken out, and the filled
// files that named them), never typed by hand.
type capturedDeclaration struct {
Builds map[string]string `json:"builds"`
Resources []json.RawMessage `json:"resources"`
Filled []string `json:"filled_files"`
Provisioner string `json:"provisioner_account"`
}
func readCaptured(t *testing.T, node string) capturedDeclaration {
t.Helper()
f, err := os.Open("testdata/declaration-" + node + ".json.gz")
if err != nil {
t.Fatal(err)
}
defer f.Close()
z, err := gzip.NewReader(f)
if err != nil {
t.Fatal(err)
}
raw, err := io.ReadAll(z)
if err != nil {
t.Fatal(err)
}
var c capturedDeclaration
if err := json.Unmarshal(raw, &c); err != nil {
t.Fatal(err)
}
return c
}
// fpOfBuild stands for each module's fingerprint in a captured declaration, which names builds, not fingerprints.
func fpOfBuild(build string) string {
var b strings.Builder
for b.Len() < 64 {
b.WriteString(fmt.Sprintf("%x", []byte(build)))
}
return SourcePrefix2 + b.String()[:64]
}
func splitCaptured(t *testing.T, node string, c capturedDeclaration) map[string]Composed {
t.Helper()
fps := map[string]string{}
for m, b := range c.Builds {
fps[m] = fpOfBuild(b)
}
// mesh-wireguard's resources are composed with no build of it named; its fingerprint is the indexer's, as every
// module's is, and stood in for here by its name.
for _, r := range c.Resources {
var head struct {
Module string `json:"module"`
}
_ = json.Unmarshal(r, &head)
if _, ok := fps[head.Module]; head.Module != "" && !ok {
fps[head.Module] = fpOfBuild(head.Module)
}
}
split, err := SplitComposed(node, c.Resources, fps, c.Filled, NodeOwn{Provisioner: c.Provisioner})
if err != nil {
t.Fatalf("SplitComposed(%s): %v", node, err)
}
return split
}
func canonical(t *testing.T, raw json.RawMessage) string {
t.Helper()
var v any
d := json.NewDecoder(bytes.NewReader(raw))
d.UseNumber()
if err := d.Decode(&v); err != nil {
t.Fatal(err)
}
out, _ := json.Marshal(v)
return string(out)
}
// 1943/AC2: each module's slice of a captured declaration of each node, and the node's own resources, round-trip
// through the composition's encoding with the same digest; together they are the declaration's resources, each once,
// each in the composition of the module it says.
func TestCompositionsOfACapturedDeclaration(t *testing.T) {
t.Parallel()
for _, node := range []string{"ace", "g14"} {
t.Run(node, func(t *testing.T) {
t.Parallel()
oneCapturedDeclaration(t, node)
})
}
}
func oneCapturedDeclaration(t *testing.T, node string) {
{
c := readCaptured(t, node)
split := splitCaptured(t, node, c)
modules := map[string]bool{}
for m := range c.Builds {
modules[m] = true
}
var union []string
for key, comp := range split {
n, module, fp, isNode, err := ParseComposedKey(key)
if err != nil || n != node || comp.Node != node || comp.Module != module || comp.FP != fp {
t.Fatalf("%s: key %q holds %s/%s/%s (%v)", node, key, comp.Node, comp.Module, comp.FP, err)
}
if !isNode && !modules[module] && module != "mesh-wireguard" {
t.Errorf("%s: a composition of %s, which the declaration names no build of", node, module)
}
raw, err := EncodeComposed(comp)
if err != nil {
t.Fatalf("%s: EncodeComposed(%s): %v", node, key, err)
}
back, err := DecodeComposed(raw)
if err != nil {
t.Fatalf("%s: DecodeComposed(%s): %v", node, key, err)
}
if !SameComposed(back, comp) {
t.Errorf("%s: %s did not round-trip", node, key)
}
// SameComposed encodes both, so the digest of what was read back, a hash of its encoding, is the one written.
if d, err := ComposedDigestOf(back); err != nil || !digestForm.MatchString(d) {
t.Errorf("%s: %s digests %q (%v)", node, key, d, err)
}
for _, r := range back.Resources {
var head struct {
Module string `json:"module"`
}
_ = json.Unmarshal(r, &head)
if head.Module != module {
t.Errorf("%s: %s holds a resource of %q", node, key, head.Module)
}
union = append(union, canonical(t, r))
}
}
var want []string
for _, r := range c.Resources {
want = append(want, canonical(t, r))
}
sort.Strings(union)
sort.Strings(want)
if strings.Join(union, "\n") != strings.Join(want, "\n") {
t.Errorf("%s: the compositions hold %d resources, the declaration %d, or they differ", node, len(union), len(want))
}
}
}
// 1943/AC2: one changed byte of a resource is another digest; the same composition written with its keys in another
// order is the same digest.
func TestComposedDigestFollowsTheContent(t *testing.T) {
c := readCaptured(t, "g14")
split := splitCaptured(t, "g14", c)
var comp Composed
for _, v := range split {
if v.Module != "" && len(v.Resources) > 1 {
comp = v
break
}
}
d1, _ := ComposedDigestOf(comp)
var m map[string]json.RawMessage
_ = json.Unmarshal(comp.Resources[0], &m)
var keys []string
for k := range m {
keys = append(keys, k)
}
sort.Sort(sort.Reverse(sort.StringSlice(keys)))
var parts []string
for _, k := range keys {
parts = append(parts, fmt.Sprintf("%q : %s", k, m[k]))
}
reordered := json.RawMessage("{ " + strings.Join(parts, ",\n ") + " }")
shuffled := comp
shuffled.Resources = append([]json.RawMessage{reordered}, comp.Resources[1:]...)
if d2, _ := ComposedDigestOf(shuffled); d2 != d1 {
t.Errorf("the same resources in another order and spacing digest %s and %s", d1, d2)
}
changed := comp
changed.Resources = append([]json.RawMessage{json.RawMessage(strings.Replace(canonical(t, comp.Resources[0]), `"id":"`, `"id":"x`, 1))}, comp.Resources[1:]...)
if d3, _ := ComposedDigestOf(changed); d3 == d1 {
t.Errorf("a changed resource kept its digest")
}
revised := comp
revised.Data = !comp.Data
if d4, _ := ComposedDigestOf(revised); d4 == d1 {
t.Errorf("a changed data flag kept its digest")
}
}
// 1943/AC2: a file over 64 KiB is carried by digest and size, its content handed back to be put in the node's object
// bucket; a composition that carries such a file whole is refused.
func TestAComposedFileOver64KiBIsCarriedByDigest(t *testing.T) {
c := readCaptured(t, "g14")
split := splitCaptured(t, "g14", c)
var comp Composed
var at int
for _, v := range split {
for i, r := range v.Resources {
var f map[string]any
_ = json.Unmarshal(r, &f)
if f["type"] == "file" && v.Module != "" && f["bytes"] == nil && f["into"] == nil {
comp, at = v, i
break
}
}
if comp.Module != "" {
break
}
}
var f map[string]any
_ = json.Unmarshal(comp.Resources[at], &f)
// Grown past the threshold from a captured file: no captured file of this node is over 64 KiB.
body := strings.Repeat("x", declaration.FileReferenceThreshold+1)
f["content"] = body
big, _ := json.Marshal(f)
comp.Resources = append([]json.RawMessage{}, comp.Resources...)
comp.Resources[at] = big
if _, err := EncodeComposed(comp); err == nil || !strings.Contains(err.Error(), "by digest") {
t.Fatalf("a file over 64 KiB carried whole = %v, want refused", err)
}
byDigest, objects, err := comp.ByDigest()
if err != nil {
t.Fatal(err)
}
var ref map[string]any
_ = json.Unmarshal(byDigest.Resources[at], &ref)
digest := declaration.DigestOf([]byte(body))
if ref["digest"] != digest || ref["size"] != float64(len(body)) || ref["content"] != nil {
t.Errorf("the file became %v, want digest %s and size %d", ref, digest, len(body))
}
if string(objects[digest]) != body || len(objects) != 1 {
t.Errorf("the objects handed back are %d, want the file's content under its digest", len(objects))
}
if _, err := EncodeComposed(byDigest); err != nil {
t.Errorf("the composition by digest is refused: %v", err)
}
}
// 1943/AC2: an unknown version is refused as one; a module's composition carrying the node's own parts, a resource of
// another module, a resource its kind refuses, or no fingerprint, is refused, naming why.
func TestComposedRefusals(t *testing.T) {
fp := fp2of('a')
file := json.RawMessage(`{"id":"m.f","type":"file","module":"m","path":"/etc/m","content":"x"}`)
ok := Composed{Node: "ace", Module: "m", FP: fp, Resources: []json.RawMessage{file}}
raw, err := EncodeComposed(ok)
if err != nil {
t.Fatal(err)
}
if _, err := DecodeComposed(bytes.Replace(raw, []byte(`"v":1`), []byte(`"v":2`), 1)); !errors.Is(err, ErrUnknownVersion) {
t.Errorf("version 2 = %v, want ErrUnknownVersion", err)
}
refusals := []struct {
c Composed
why string
}{
{Composed{Node: "ace", Module: "m", Resources: []json.RawMessage{file}}, "fingerprint"},
{Composed{Node: "", Module: "m", FP: fp, Resources: []json.RawMessage{file}}, "node name is empty"},
{Composed{Node: "ace", Module: "other", FP: fp, Resources: []json.RawMessage{file}}, `of "m"`},
{Composed{Node: "ace", Module: "m", FP: fp, Resources: []json.RawMessage{file}, Own: NodeOwn{Provisioner: "p"}}, "node's own"},
{Composed{Node: "ace", FP: fp, Resources: []json.RawMessage{}}, "names no fingerprint"},
{Composed{Node: "ace", Resources: []json.RawMessage{file}}, `of "m"`},
{Composed{Node: "ace", Module: "m", FP: fp, Resources: []json.RawMessage{json.RawMessage(`{"id":"m.f","type":"file","module":"m","path":"/etc/m","content":"x","image":"y"}`)}}, `"image"`},
{Composed{Node: "ace", Module: "m", FP: fp, Resources: []json.RawMessage{json.RawMessage(`{"id":"m.a","type":"action","module":"m"}`)}}, "action"},
}
for _, r := range refusals {
if _, err := EncodeComposed(r.c); err == nil || !strings.Contains(err.Error(), r.why) {
t.Errorf("EncodeComposed(%s/%s) = %v, want a refusal saying %q", r.c.Node, r.c.Module, err, r.why)
}
}
}
// 1943/AC2 (issue 1150): a module whose name holds a dot keys its have, its want and its composition escaped, and the
// parsers give its name back; a name with a wildcard, whitespace or an escape of its own is still refused.
func TestADottedModuleKeysEscaped(t *testing.T) {
fp := fp2of('f')
have, err := HaveKey("ace", "novox.be")
if err != nil || have != "ace.novox=2ebe" {
t.Fatalf("HaveKey = %q, %v", have, err)
}
if n, m, isNode, err := ParseHaveKey(have); err != nil || n != "ace" || m != "novox.be" || isNode {
t.Errorf("ParseHaveKey(%q) = %q, %q, %v, %v", have, n, m, isNode, err)
}
want, err := WantKey("ace", "novox.be")
if err != nil || want != "want.ace.novox=2ebe" {
t.Fatalf("WantKey = %q, %v", want, err)
}
if n, m, err := ParseWantKey(want); err != nil || n != "ace" || m != "novox.be" {
t.Errorf("ParseWantKey(%q) = %q, %q, %v", want, n, m, err)
}
comp, err := ComposedKey("ace", "novox.be", fp)
if err != nil || comp != "ace.novox=2ebe."+strings.Repeat("f", 64) {
t.Fatalf("ComposedKey = %q, %v", comp, err)
}
if n, m, back, _, err := ParseComposedKey(comp); err != nil || n != "ace" || m != "novox.be" || back != fp {
t.Errorf("ParseComposedKey(%q) = %q, %q, %q, %v", comp, n, m, back, err)
}
for _, bad := range []string{"m*", "m x", "m=2e", "m>", ".m", "m.", "a..b"} {
if _, err := HaveKey("ace", bad); err == nil {
t.Errorf("HaveKey(ace, %q) accepted", bad)
}
}
for _, bad := range []string{"ace.m=2", "ace.m=2E", "ace.m=41", "ace.=6eode"} {
if _, _, _, err := ParseHaveKey(bad); err == nil {
t.Errorf("ParseHaveKey(%q) accepted", bad)
}
}
}
// The node's own composition carries the filled paths of the files moved into it, the files of a module with no
// fingerprint of the new form: the provisioner fills them all the same (issue 1945, live on 2026-10-25: the paths were
// dropped and the node-engine wrote those files empty). It still holds no module's data.
func TestTheNodesOwnCompositionCarriesTheFilledPathsOfItsFiles(t *testing.T) {
filled := json.RawMessage(`{"id":"power.unit","type":"file","path":"/etc/systemd/system/p.service","mode":"0644","owner-at-runtime":"owned"}`)
own := Composed{Node: "ace", Resources: []json.RawMessage{filled}, Filled: []string{"/etc/systemd/system/p.service"},
Own: NodeOwn{Provisioner: "mesh-provisioner-client"}}
raw, err := EncodeComposed(own)
if err != nil {
t.Fatalf("the node's own composition with a filled file: %v", err)
}
back, err := DecodeComposed(raw)
if err != nil || len(back.Filled) != 1 {
t.Fatalf("read back: %v %v", back.Filled, err)
}
if _, err := EncodeComposed(Composed{Node: "ace", Data: true, Resources: []json.RawMessage{filled},
Filled: own.Filled}); err == nil {
t.Error("the node's own composition was let hold a module's data")
}
}
// A file the provisioner fills says no content: in a composition that does not say it is filled, the node-engine would
// write it empty. Such a composition is refused, in the node's own as in a module's (issue 1945).
func TestAFileWithNoContentThatIsNotFilledIsRefused(t *testing.T) {
fp := fp2of('a')
for _, c := range []Composed{
{Node: "ace", Resources: []json.RawMessage{json.RawMessage(
`{"id":"power.unit","type":"file","path":"/etc/p","mode":"0644","owner-at-runtime":"owned"}`)}},
{Node: "ace", Module: "m", FP: fp, Resources: []json.RawMessage{json.RawMessage(
`{"id":"m.f","type":"file","module":"m","path":"/etc/m","mode":"0644","owner-at-runtime":"owned"}`)}},
} {
if _, err := EncodeComposed(c); err == nil || !strings.Contains(err.Error(), "empty") {
t.Errorf("%s/%s: %v, want a refusal saying it would be written empty", c.Node, c.Module, err)
}
}
// A file that says its content, even an empty one, is written as it says.
if _, err := EncodeComposed(Composed{Node: "ace", Resources: []json.RawMessage{json.RawMessage(
`{"id":"x.f","type":"file","path":"/etc/x","mode":"0644","content":""}`)}}); err != nil {
t.Errorf("a file that says it is empty: %v", err)
}
}
+72 -22
View File
@@ -11,7 +11,8 @@
// `node`. A presence key is `<node>` for a node and `part.<part>.<instance>` for a part's instance, so a node may
// not be named `part`. A pulse key is `<bucket>.<key>`: the full name of the bucket watched, then that bucket's key,
// which may hold dots, so it splits at the first dot. A name is refused when it is empty or holds a dot, a wildcard,
// whitespace or a character a bus key refuses, and the error names it.
// whitespace or a character a bus key refuses, and the error names it; but a module's name may hold dots between its
// parts (`novox.be`), and keys with each dot escaped as EscapeName writes it (`novox=2ebe`, issue 1150).
//
// **Values** are JSON of under about 500 bytes, each carrying its version in `v`. A reader refuses a value whose
// version is missing or one it does not know, with an error that is ErrUnknownVersion (research 063 section 05
@@ -35,7 +36,7 @@
// builder's build keys (builder.go) and the checker's verdicts and kept verdicts (checker.go). A manifests value is the
// one exception to the size above: a manifest is up to about 21 KB, whole, byte for byte.
//
// Standard library only, and this module's own asks package.
// Standard library only, and this module's own asks and declaration packages (a composition holds declaration resources).
package delivery
import (
@@ -89,6 +90,10 @@ const (
// for which it had written a have key; the key is never deleted (issue 1369, ADR 0408 §4), so a reader can tell
// that the module stopped.
PhaseRemoved Phase = "removed"
// PhaseHeld is a module whose backup before an apply failed: it keeps running what it has, nothing of the new
// composition is applied, and the node waits for the answer in its want (novox/hq ADR 0539 clause 16). Detail
// carries the backup's own words.
PhaseHeld Phase = "held"
)
// Health is what the module's own check says.
@@ -164,13 +169,45 @@ func HaveKey(node, module string) (string, error) {
if err := checkName("node", node); err != nil {
return "", err
}
if err := checkName("module", module); err != nil {
part, err := moduleKeyPart(module)
if err != nil {
return "", err
}
if module == "node" {
return "", fmt.Errorf(`delivery: the module name "node" is refused: %s.node is the node's own key`, node)
}
return node + "." + module, nil
return node + "." + part, nil
}
var moduleNameOK = regexp.MustCompile(`^[-/_a-zA-Z0-9]+(\.[-/_a-zA-Z0-9]+)*$`)
// moduleKeyPart is a module's name as a key part (issue 1150): a name may hold a dot (`novox.be`), which a key part
// may not, so a dot is escaped as EscapeName writes it (`novox=2ebe`); a name of letters, digits, `-`, `/` and `_` keys
// as it is. A name that is empty, or holds a wildcard, whitespace, `=`, or an empty part between dots, is refused.
func moduleKeyPart(module string) (string, error) {
if module == "" {
return "", fmt.Errorf("delivery: the module name is empty")
}
if !moduleNameOK.MatchString(module) {
return "", fmt.Errorf("delivery: the module name %q holds a wildcard, whitespace, '=', an empty part between dots or a character a key refuses", module)
}
return EscapeName(module), nil
}
// parseModuleKeyPart is the inverse of moduleKeyPart; it refuses a part that is not the one moduleKeyPart writes for
// its name (an escape of a character that keys as it is, an upper-case escape, an escape of a refused character).
func parseModuleKeyPart(part string) (string, error) {
if err := checkName("module", part); err != nil {
return "", err
}
module, err := UnescapeName(part)
if err != nil {
return "", err
}
if back, err := moduleKeyPart(module); err != nil || back != part {
return "", fmt.Errorf("delivery: the module part %q is not a module name as a key writes it", part)
}
return module, nil
}
// HaveNodeKey is the key, in HaveBucket, of a node's own facts: `<node>.node`.
@@ -190,12 +227,13 @@ func ParseHaveKey(key string) (node, module string, isNode bool, err error) {
if err := checkName("node", node); err != nil {
return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
}
if err := checkName("module", module); err != nil {
return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
}
if module == "node" {
return node, "", true, nil
}
module, err = parseModuleKeyPart(module)
if err != nil {
return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
}
return node, module, false, nil
}
@@ -282,24 +320,31 @@ func PulseTTL(moveAge time.Duration) (ttl time.Duration, stalled bool) {
// Have is a module's state on a node, written by that node's node-engine under HaveKey. FP is the module's
// fingerprint once the indexer exists (design 47 release 3); until then Build is the build the node was sent, and
// the view compares by build.
//
// Composed is the digest of the composition the node runs for the module (novox/hq ADR 0539 clause 9): the
// controller judges a move against it (clause 14), and the assigner reads it to know a want was applied.
type Have struct {
V int `json:"v"`
FP string `json:"fp,omitempty"`
Build string `json:"build,omitempty"`
Phase Phase `json:"phase"`
Health Health `json:"health"`
Detail string `json:"detail,omitempty"`
V int `json:"v"`
FP string `json:"fp,omitempty"`
Composed string `json:"composed,omitempty"`
Build string `json:"build,omitempty"`
Phase Phase `json:"phase"`
Health Health `json:"health"`
Detail string `json:"detail,omitempty"`
}
// HaveNode is a node's own facts, written under HaveNodeKey. Each field is the node-engine's own JSON, carried whole:
// this package names the facts and does not type them.
//
// Composed is the digest of the node's own composition it runs (ADR 0539 clause 9).
type HaveNode struct {
V int `json:"v"`
Windows json.RawMessage `json:"windows,omitempty"`
Busy json.RawMessage `json:"busy,omitempty"`
Strays json.RawMessage `json:"strays,omitempty"`
Filters json.RawMessage `json:"filters,omitempty"`
Profile json.RawMessage `json:"profile,omitempty"`
V int `json:"v"`
Composed string `json:"composed,omitempty"`
Windows json.RawMessage `json:"windows,omitempty"`
Busy json.RawMessage `json:"busy,omitempty"`
Strays json.RawMessage `json:"strays,omitempty"`
Filters json.RawMessage `json:"filters,omitempty"`
Profile json.RawMessage `json:"profile,omitempty"`
}
// Presence is whether a node or a part's instance is online; the entry's own time says since when. The watchman
@@ -347,9 +392,14 @@ func decode(data []byte, into any) error {
func (h Have) check() error {
switch h.Phase {
case PhaseBackingUp, PhaseApplying, PhaseChecking, PhaseRunning, PhaseFailed, PhaseRemoved:
case PhaseBackingUp, PhaseApplying, PhaseChecking, PhaseRunning, PhaseFailed, PhaseRemoved, PhaseHeld:
default:
return fmt.Errorf("delivery: phase %q is not backing-up, applying, checking, running, failed or removed", h.Phase)
return fmt.Errorf("delivery: phase %q is not backing-up, applying, checking, running, failed, removed or held", h.Phase)
}
if h.Composed != "" {
if err := checkDigest(h.Composed, "the have's composition digest"); err != nil {
return err
}
}
switch h.Health {
case HealthOK, HealthFailed, HealthWaits, HealthUnknown:
@@ -424,7 +474,7 @@ func SameHave(a, b Have) bool { return a == b }
// SameHaveNode says whether two node facts are the same value, comparing each carried JSON without its spacing.
func SameHaveNode(a, b HaveNode) bool {
return a.V == b.V && sameJSON(a.Windows, b.Windows) && sameJSON(a.Busy, b.Busy) &&
return a.V == b.V && a.Composed == b.Composed && sameJSON(a.Windows, b.Windows) && sameJSON(a.Busy, b.Busy) &&
sameJSON(a.Strays, b.Strays) && sameJSON(a.Filters, b.Filters) && sameJSON(a.Profile, b.Profile)
}
+17
View File
@@ -73,3 +73,20 @@ func DecodeControllerGrant(data []byte) (ControllerGrant, error) {
// SameControllerGrant says whether two grants are the same value.
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
}
+148
View File
@@ -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
}
+6 -2
View File
@@ -535,9 +535,13 @@ func ParseAskStepKey(key string) (node, askID string, err error) {
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
// 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) {
if err := checkName("node", node); err != nil {
return "", err
Binary file not shown.
Binary file not shown.
+71 -11
View File
@@ -32,19 +32,27 @@ const (
AskKindHold = "hold"
// AskKindRestore is the kind of the assigner's ask about a data module set back.
AskKindRestore = "restore"
// AskKindBackup is the kind of the assigner's ask about a backup before an apply that failed: back up again, or go
// on without one (novox/hq ADR 0539 clause 16).
AskKindBackup = "backup"
// AskKindRecord is the kind of the assigner's ask about a move a person must allow, for a module of policy record:
// a composition judged to recreate a container or move data, or one that cannot be judged (ADR 0539 clause 14).
AskKindRecord = "record"
viewWord = "view"
)
// AskKey is an ask's key read whole: its Kind, and the parts its kind has: a merge's Group and heads Hash, a hold's
// Module and fingerprint FP, a restore's Node, Module and FP.
// Module and fingerprint FP, a restore's and a backup's Node, Module and FP, a record's Node, Module and the digest
// Composed of the composition it would allow.
type AskKey struct {
Kind string
Group string
Hash string
Node string
Module string
FP string
Kind string
Group string
Hash string
Node string
Module string
FP string
Composed string
}
// AskNeedsProof says whether an ask of this kind must prove who answered it, so the view may not answer it: the
@@ -53,10 +61,11 @@ func AskNeedsProof(kind string) (bool, error) {
switch kind {
case AskKindMerge:
return true, nil
case AskKindHold, AskKindRestore:
case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord:
return false, nil
}
return false, fmt.Errorf("delivery: %q is not a kind of ask, %s, %s or %s", kind, AskKindMerge, AskKindHold, AskKindRestore)
return false, fmt.Errorf("delivery: %q is not a kind of ask, %s, %s, %s, %s or %s", kind, AskKindMerge, AskKindHold,
AskKindRestore, AskKindBackup, AskKindRecord)
}
// HoldAskKey is the key, in AssignerAsksBucket, of the ask about a module's held fingerprint: `hold.<module>.<hex>`.
@@ -81,7 +90,32 @@ func RestoreAskKey(node, module, fingerprint string) (string, error) {
return AskKindRestore + "." + node + "." + key, nil
}
// ParseAnyAskKey reads an ask key of any kind (MergeAskKey, HoldAskKey, RestoreAskKey); it refuses an unknown kind
// BackupAskKey is the key, in AssignerAsksBucket, of the ask about a module's failed backup on a node before it applies
// a fingerprint: `backup.<node>.<module>.<hex>`.
func BackupAskKey(node, module, fingerprint string) (string, error) {
key, err := ComposedKey(node, module, fingerprint)
if err != nil {
return "", err
}
return AskKindBackup + "." + key, nil
}
// RecordAskKey is the key, in AssignerAsksBucket, of the ask about a move of a module on a node to the composition
// whose digest it names: `record.<node>.<module>.<digest hex>`, so a new revision at the same fingerprint is asked
// again (ADR 0539 clause 14).
func RecordAskKey(node, module, composed string) (string, error) {
have, err := HaveKey(node, module)
if err != nil {
return "", err
}
if err := checkDigest(composed, "a record ask's composition digest"); err != nil {
return "", err
}
return AskKindRecord + "." + have + "." + strings.TrimPrefix(composed, "sha256:"), nil
}
// ParseAnyAskKey reads an ask key of any kind (MergeAskKey, HoldAskKey, RestoreAskKey, BackupAskKey, RecordAskKey); it
// refuses an unknown kind
// and a key whose parts do not fit its kind, naming the key.
func ParseAnyAskKey(key string) (AskKey, error) {
kind, rest, _ := strings.Cut(key, ".")
@@ -114,6 +148,26 @@ func ParseAnyAskKey(key string) (AskKey, error) {
return refused(err)
}
return AskKey{Kind: kind, Node: node, Module: module, FP: fp}, nil
case AskKindBackup:
node, module, fp, isNode, err := ParseComposedKey(rest)
if err != nil || isNode {
return refused(fmt.Errorf("a backup ask is backup.<node>.<module>.<fingerprint hex>"))
}
return AskKey{Kind: kind, Node: node, Module: module, FP: fp}, nil
case AskKindRecord:
parts := strings.Split(rest, ".")
if len(parts) != 3 {
return refused(fmt.Errorf("a record ask is record.<node>.<module>.<digest hex>"))
}
node, module, isNode, err := ParseHaveKey(parts[0] + "." + parts[1])
if err != nil || isNode {
return refused(fmt.Errorf("a record ask is record.<node>.<module>.<digest hex>"))
}
composed := "sha256:" + parts[2]
if err := checkDigest(composed, "a record ask's composition digest"); err != nil {
return refused(err)
}
return AskKey{Kind: kind, Node: node, Module: module, Composed: composed}, nil
}
_, err := AskNeedsProof(kind)
return refused(err)
@@ -229,9 +283,15 @@ func ViewAnswerFits(a ViewAnswer, ask AssignerAsk) error {
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)
}
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)
}
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
}
+7
View File
@@ -158,4 +158,11 @@ func TestViewAnswerFits(t *testing.T) {
if err := ViewAnswerFits(ViewAnswer{Asking: first.Ask.ID, Option: "restore"}, first); err == nil {
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)
}
}
+176
View File
@@ -0,0 +1,176 @@
package delivery
import (
"strings"
"testing"
)
func hashOf(c byte) string { return "sha256:" + digestOf(c) }
// 1943/AC3: a want naming its fingerprint and the digest of the composition it allows round-trips, and so does one
// carrying the answer to a failed backup; a want naming a fingerprint without a digest, a digest not of the sha256
// form, a removed want naming a digest, an unknown backup answer and a rule past the seventh are refused.
func TestAWantNamesItsComposition(t *testing.T) {
for _, w := range []Want{
{FP: fp2of('a'), Composed: hashOf('1')},
{FP: fp2of('a'), Composed: hashOf('1'), Backup: BackupAgain},
{FP: fp2of('a'), Composed: hashOf('1'), Backup: BackupSkip, Words: "the operator: go on without a backup"},
{FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: RuleComposed, Words: "no composition of b for ace yet"},
{Removed: true},
} {
raw, err := EncodeWant(w)
if err != nil {
t.Fatalf("EncodeWant(%+v): %v", w, err)
}
back, err := DecodeWant(raw)
w.V = Version
if err != nil || !SameWant(back, w) {
t.Errorf("want %+v came back %+v, %v", w, back, err)
}
}
refusals := []struct {
w Want
why string
}{
{Want{FP: fp2of('a')}, "names no composition"},
{Want{FP: fp2of('a'), Composed: "sha256:12"}, "not sha256"},
{Want{Removed: true, Composed: hashOf('1')}, "removed want names no"},
{Want{FP: fp2of('a'), Composed: hashOf('1'), Backup: "later"}, `backup answer "later"`},
{Want{Removed: true, Backup: BackupAgain}, "removed want names no"},
{Want{FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: 8}, "rule 8"},
}
for _, r := range refusals {
if _, err := EncodeWant(r.w); err == nil || !strings.Contains(err.Error(), r.why) {
t.Errorf("EncodeWant(%+v) = %v, want a refusal saying %q", r.w, err, r.why)
}
}
if _, err := DecodeWant([]byte(`{"v":1,"fp":"` + fp2of('a') + `"}`)); err == nil || !strings.Contains(err.Error(), "names no composition") {
t.Errorf("a want of release 5, naming no composition, was read: %v", err)
}
if RuleComposed != 7 {
t.Errorf("RuleComposed = %d, want the seventh rule", RuleComposed)
}
}
// 1943/AC3: the node's own want is `want.<node>.node` and round-trips through its parser, while a module named node is
// still refused a want key; its value names the digest of the node's own composition and no fingerprint.
func TestTheNodesOwnWant(t *testing.T) {
key, err := WantNodeKey("ace")
if err != nil || key != "want.ace.node" {
t.Fatalf("WantNodeKey = %q, %v", key, err)
}
if node, err := ParseWantNodeKey(key); err != nil || node != "ace" {
t.Errorf("ParseWantNodeKey(%q) = %q, %v", key, node, err)
}
if _, err := WantKey("ace", "node"); err == nil {
t.Errorf(`WantKey(ace, "node") accepted`)
}
if _, _, err := ParseWantKey(key); err == nil {
t.Errorf("ParseWantKey read the node's own want as a module's")
}
for _, bad := range []string{"want.ace.mesh-mail", "want.node", "ace.node", "want.a.b.node", ""} {
if _, err := ParseWantNodeKey(bad); err == nil || !strings.Contains(err.Error(), "\""+bad+"\"") {
t.Errorf("ParseWantNodeKey(%q) = %v, want a refusal naming it", bad, err)
}
}
if _, err := WantNodeKey(""); err == nil {
t.Errorf("WantNodeKey(\"\") accepted")
}
w := NodeWant{Composed: hashOf('2'), Words: "the node's own composition"}
raw, err := EncodeNodeWant(w)
if err != nil {
t.Fatal(err)
}
back, err := DecodeNodeWant(raw)
w.V = Version
if err != nil || back != w {
t.Errorf("node want came back %+v, %v", back, err)
}
for _, bad := range []NodeWant{{}, {Composed: "sha256:1"}} {
if _, err := EncodeNodeWant(bad); err == nil {
t.Errorf("EncodeNodeWant(%+v) accepted", bad)
}
}
if _, err := DecodeNodeWant([]byte(`{"v":1,"composed":"` + hashOf('2') + `","fp":"` + fp2of('a') + `"}`)); err == nil {
t.Errorf("a node want naming a fingerprint was read")
}
}
// 1943/AC3: a have held after a failed backup, naming the composition it runs, round-trips; so does the node's own have
// naming its composition; a have's digest not of the sha256 form is refused.
func TestAHaveNamesItsComposition(t *testing.T) {
for _, h := range []Have{
{FP: fp2of('a'), Composed: hashOf('3'), Phase: PhaseHeld, Health: HealthOK, Detail: "the backup failed: no space left"},
{FP: fp2of('a'), Composed: hashOf('3'), Phase: PhaseRunning, Health: HealthOK},
} {
raw, err := EncodeHave(h)
if err != nil {
t.Fatalf("EncodeHave(%+v): %v", h, err)
}
back, err := DecodeHave(raw)
h.V = Version
if err != nil || !SameHave(back, h) {
t.Errorf("have %+v came back %+v, %v", h, back, err)
}
}
if _, err := EncodeHave(Have{Composed: "sha256:x", Phase: PhaseRunning, Health: HealthOK}); err == nil || !strings.Contains(err.Error(), "not sha256") {
t.Errorf("a have naming a bad digest = %v", err)
}
n := HaveNode{Composed: hashOf('4')}
raw, err := EncodeHaveNode(n)
if err != nil {
t.Fatal(err)
}
back, err := DecodeHaveNode(raw)
n.V = Version
if err != nil || !SameHaveNode(back, n) {
t.Errorf("node have came back %+v, %v", back, err)
}
if SameHaveNode(n, HaveNode{V: Version, Composed: hashOf('5')}) {
t.Errorf("node haves naming two compositions are the same")
}
}
// 1943/AC3: the assigner's backup ask (back up again, or go on without one, ADR 0539 clause 16) and its record ask (a
// move a person must allow, clause 14) key by node, module and fingerprint or composition, round-trip through
// ParseAnyAskKey, and the view may answer both.
func TestTheBackupAndRecordAsks(t *testing.T) {
fp, digest := fp2of('a'), hashOf('6')
backup, err := BackupAskKey("ace", "mesh-mail", fp)
if err != nil || backup != "backup.ace.mesh-mail."+strings.Repeat("a", 64) {
t.Fatalf("BackupAskKey = %q, %v", backup, err)
}
record, err := RecordAskKey("ace", "mesh-mail", digest)
if err != nil || record != "record.ace.mesh-mail."+strings.Repeat("6", 64) {
t.Fatalf("RecordAskKey = %q, %v", record, err)
}
if got, err := ParseAnyAskKey(backup); err != nil || got != (AskKey{Kind: AskKindBackup, Node: "ace", Module: "mesh-mail", FP: fp}) {
t.Errorf("ParseAnyAskKey(%q) = %+v, %v", backup, got, err)
}
if got, err := ParseAnyAskKey(record); err != nil || got != (AskKey{Kind: AskKindRecord, Node: "ace", Module: "mesh-mail", Composed: digest}) {
t.Errorf("ParseAnyAskKey(%q) = %+v, %v", record, got, err)
}
for _, kind := range []string{AskKindBackup, AskKindRecord} {
if proof, err := AskNeedsProof(kind); err != nil || proof {
t.Errorf("AskNeedsProof(%s) = %v, %v; the view may answer it", kind, proof, err)
}
}
for _, k := range []string{backup, record} {
if _, err := ViewAnswerKey(k); err != nil {
t.Errorf("the view may not answer %s: %v", k, err)
}
}
for _, bad := range [][3]string{{"", "mesh-mail", fp}, {"ace", "node", fp}, {"ace", "mesh-mail", "src2:1"}} {
if _, err := BackupAskKey(bad[0], bad[1], bad[2]); err == nil {
t.Errorf("BackupAskKey(%v) accepted", bad)
}
}
if _, err := RecordAskKey("ace", "mesh-mail", "sha256:1"); err == nil {
t.Errorf("a record ask naming a bad digest was keyed")
}
for _, bad := range []string{"backup.ace.mesh-mail", "record.ace.mesh-mail.12", "record.ace." + strings.Repeat("6", 64)} {
if _, err := ParseAnyAskKey(bad); err == nil {
t.Errorf("ParseAnyAskKey(%q) accepted", bad)
}
}
}
+32
View File
@@ -1,11 +1,16 @@
package process
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"strings"
"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,
@@ -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
// in the error even when it is the field present and another is missing.
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 {
field string
values map[string]any
@@ -82,3 +109,8 @@ func writeAccount(t *testing.T, fields map[string]any) string {
}
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 }
+55
View File
@@ -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)
}
+8
View File
@@ -154,8 +154,15 @@ const CodeChanged = -32010
var (
current *channel
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
// module the runtime named in MESH_SERVED_MODULE.
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()
current = c
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
// it is asked from beside the loop that reads the answer.
go func() { _ = announce1() }()
+21
View File
@@ -16,6 +16,22 @@ func TestABundleEmitsThroughTheRuntimeAndServesAsItsModule(t *testing.T) {
t.Setenv(ServedAs, "chime")
t.Setenv("MESH_MODULE", "chime")
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()
outR, outW := io.Pipe()
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" {
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{}}})
asked := next()
if asked["method"] != "mesh/publish" {