novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

The composition pass never waits on marking haves, and the compose verb answers while a pass runs

Browse Source 
The marks of the haves (a refusal, the words a module keeps) read and write each module's have in turn, and each write
walks every have: on a slow bus the pass waited minutes on them (shanks, 2026-11-05 22:22). They now run off the pass,
newest word per kind, each bounded; an ask of compose during a pass is answered at once and queues one more pass.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 22:33:21 +01:00
parent e73aae3686
commit c1b8b9d115
5 changed files with 222 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/marker.go
+76
View File
@@ -0,0 +1,76 @@
package main
import (
"context"
"sync"
"time"
)
// markWithin bounds one marking of the haves (what the composer tells its agent: a refusal, the words it keeps). The marks
// read and write each module's have over the bus one after the other; a bus that answers slowly or not at all must hold
// the marking, never the composition pass (issue 1965, shanks 2026-11-05 22:22: a pass waited minutes on them).
var markWithin = 45 * time.Second
// marker runs the composer's marks of the haves off the composition pass: set returns at once, one worker runs what was
// set last under each name, in the order the names were first set, each within markWithin. A mark set while one runs
// replaces the pending mark of its name, so the haves are marked once for the newest word, not once for every pass.
type marker struct {
run context.Context
mu sync.Mutex
pending map[string]func(context.Context)
order []string
working bool
idle *sync.Cond
}
func newMarker(run context.Context) *marker {
m := &marker{run: run, pending: map[string]func(context.Context){}}
m.idle = sync.NewCond(&m.mu)
return m
}
// set takes the mark name should carry; it never waits on the bus.
func (m *marker) set(name string, mark func(context.Context)) {
m.mu.Lock()
defer m.mu.Unlock()
if _, queued := m.pending[name]; !queued {
m.order = append(m.order, name)
}
m.pending[name] = mark
if m.working {
return
}
m.working = true
go m.work()
}
func (m *marker) work() {
for {
m.mu.Lock()
if len(m.order) == 0 || m.run.Err() != nil {
m.order, m.pending = nil, map[string]func(context.Context){}
m.working = false
m.idle.Broadcast()
m.mu.Unlock()
return
}
name := m.order[0]
m.order = m.order[1:]
mark := m.pending[name]
delete(m.pending, name)
m.mu.Unlock()
ctx, cancel := context.WithTimeout(m.run, markWithin)
mark(ctx)
cancel()
}
}
// wait returns once no mark is pending or running (for tests).
func (m *marker) wait() {
m.mu.Lock()
defer m.mu.Unlock()
for m.working {
m.idle.Wait()
}
}
cmd/mesh-host/marker_test.go
+47
View File
@@ -0,0 +1,47 @@
package main
import (
"context"
"testing"
"time"
)
// Issue 1965 (shanks 2026-11-05 22:22): the marks of the haves never hold the composition pass. A mark that stalls on the
// bus is cut at markWithin, a mark set while one runs replaces the pending one of its name, and set itself never waits.
func TestMarksOfTheHavesNeverHoldThePass(t *testing.T) {
setMarkWithin(t, 30*time.Millisecond)
m := newMarker(context.Background())
var ran []string
stalled := make(chan struct{})
began := time.Now()
m.set("refused", func(ctx context.Context) { // a have write that stalls until its bound
close(stalled)
<-ctx.Done()
ran = append(ran, "refused-stalled")
})
<-stalled
// Set while the first runs: only the newest of a name is kept.
m.set("kept", func(context.Context) { ran = append(ran, "kept-old") })
m.set("kept", func(context.Context) { ran = append(ran, "kept-new") })
m.set("refused", func(context.Context) { ran = append(ran, "refused-new") })
if took := time.Since(began); took > 20*time.Millisecond {
t.Errorf("setting marks took %v; it must not wait on the stalled one", took)
}
m.wait()
want := []string{"refused-stalled", "kept-new", "refused-new"}
if len(ran) != len(want) {
t.Fatalf("ran %v; want %v", ran, want)
}
for i := range want {
if ran[i] != want[i] {
t.Fatalf("ran %v; want %v", ran, want)
}
}
}
func setMarkWithin(t *testing.T, d time.Duration) {
t.Helper()
old := markWithin
markWithin = d
t.Cleanup(func() { markWithin = old })
}
cmd/mesh-host/nodeagent.go
+5 -2
View File
@@ -586,6 +586,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
// The agent of this run alone: its composer tells it, never the agent of a run begun since (issue 1126: a pass in
// flight at the switch marked the next agent's modules failed).
own := &agentNow{}
// The marks of the haves run off the composition pass, each bounded (see marker).
marks := newMarker(run)
hooks := composerHooks{
running: func() []delivery.Composed {
if now := own.get(); now != nil {
@@ -609,13 +611,14 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
},
refused: func(why string, held map[string]string) {
if now := own.get(); now != nil && run.Err() == nil {
now.NotComposedEach(run, why, held) // a module held alone marks that module, never the others (issue 1126)
// A module held alone marks that module, never the others (issue 1126); the marking never holds the pass.
marks.set("refused", func(ctx context.Context) { now.NotComposedEach(ctx, why, held) })
}
},
// And what it keeps of a module as it runs, for a manifest that does not say it yet: on the module's have (ADR 0601 §4).
kept: func(kept map[string][]string) {
if now := own.get(); now != nil && run.Err() == nil {
now.KeptNow(run, kept)
marks.set("kept", func(ctx context.Context) { now.KeptNow(ctx, kept) })
}
},
// And the vault's authority beside the controller's, and this node's certificate asked of the vault when a
internal/selfcompose/compose_while_passing_test.go
+75
View File
@@ -0,0 +1,75 @@
package selfcompose
import (
"context"
"strings"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-host/internal/agent"
)
// Issue 1965 (shanks 2026-11-05 22:27: the verb compose gave no answer in 30 s while a pass waited on the bus): an ask
// that comes while a pass runs is answered at once, says one more pass is queued and what the running one waits on, and
// the queued pass is made when the running one ends.
func TestComposeAnsweredWhileAPassRuns(t *testing.T) {
b := newFakeBus()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
env := b.env()
hold := make(chan struct{})
var passes atomic.Int64
reached := make(chan struct{}, 8)
env.Running = func() []delivery.Composed {
passes.Add(1)
reached <- struct{}{}
if passes.Load() == 1 {
<-hold // the first pass waits on the bus
}
return nil
}
c := New(env)
out, err := c.Watch(ctx, "n1.>")
if err != nil {
t.Fatal(err)
}
have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}})
place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}})
nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true})
b.put(t, delivery.PlacementBucket, "placement.n1", place)
b.put(t, delivery.HaveBucket, "n1.node", have)
b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant)
for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket, delivery.BuilderBuildsBucket} {
b.send(bucket, agent.Update{Caught: true})
}
go func() {
for range out {
}
}()
select {
case <-reached:
case <-time.After(5 * time.Second):
t.Fatal("no pass began")
}
answered := make(chan Pass, 1)
go func() {
p, _ := c.ComposeNow(ctx)
answered <- p
}()
select {
case p := <-answered:
if !strings.Contains(p.Queued, "a pass is running") {
t.Errorf("answered %+v; want it to say a pass is running and one is queued", p)
}
case <-time.After(2 * time.Second):
t.Fatal("the verb compose gave no answer while a pass ran")
}
close(hold)
select {
case <-reached: // the queued pass
case <-time.After(5 * time.Second):
t.Fatal("the queued pass was not made when the running one ended")
}
}
internal/selfcompose/selfcompose.go
+19 -1
View File
@@ -167,6 +167,8 @@ type Composer struct {
wait time.Duration
// asked are the compose verb's asks (ComposeNow), each answered once the pass it asks is made.
asked chan chan Pass
// passing is set while a pass is made.
passing atomic.Bool
// last is how the last pass went, under mu.
last Pass
// watching says the composer's loop runs (Watch), so an ask of a pass can be answered.
@@ -216,6 +218,9 @@ type Pass struct {
Again time.Time `json:"again,omitempty"`
// Earlier says the answer is the last pass, made less than askedWithin ago, and no new one was made for the ask.
Earlier bool `json:"earlier,omitempty"`
// Queued says a pass was running when the ask came: one more is made when it ends, and this answer is not its result.
// It says what the running pass waits on and since when.
Queued string `json:"queued,omitempty"`
}
// LastHeld is what the last pass held, each module with why; nil before a pass or when it held none.
@@ -307,6 +312,16 @@ func (c *Composer) ComposeNow(ctx context.Context) (Pass, error) {
return last, nil
}
reply := make(chan Pass, 1)
if c.passing.Load() {
// A pass runs: the ask joins the one made when it ends (all asks of the time share it), and is answered now, never
// held for the pass (issue 1965: the verb gave no answer for minutes while a pass waited on the bus).
select {
case c.asked <- reply:
default: // eight asks already wait for that pass
}
last.Queued = "a pass is running and waits on " + c.waitingOn() + "; one more is made when it ends"
return last, nil
}
select {
case c.asked <- reply:
case <-ctx.Done():
@@ -496,7 +511,10 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er
continue
}
dirty = false
for _, u := range c.compose(ctx) {
c.passing.Store(true)
updates := c.compose(ctx)
c.passing.Store(false)
for _, u := range updates {
select {
case out <- u:
case <-ctx.Done():