From c1b8b9d115b96c2a2501cb3550a0841d2a4960e8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 22:33:21 +0100 Subject: [PATCH] The composition pass never waits on marking haves, and the compose verb answers while a pass runs 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 --- cmd/mesh-host/marker.go | 76 +++++++++++++++++++ cmd/mesh-host/marker_test.go | 47 ++++++++++++ cmd/mesh-host/nodeagent.go | 7 +- .../selfcompose/compose_while_passing_test.go | 75 ++++++++++++++++++ internal/selfcompose/selfcompose.go | 20 ++++- 5 files changed, 222 insertions(+), 3 deletions(-) create mode 100644 cmd/mesh-host/marker.go create mode 100644 cmd/mesh-host/marker_test.go create mode 100644 internal/selfcompose/compose_while_passing_test.go diff --git a/cmd/mesh-host/marker.go b/cmd/mesh-host/marker.go new file mode 100644 index 0000000..2f5cbd7 --- /dev/null +++ b/cmd/mesh-host/marker.go @@ -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() + } +} diff --git a/cmd/mesh-host/marker_test.go b/cmd/mesh-host/marker_test.go new file mode 100644 index 0000000..68c86e5 --- /dev/null +++ b/cmd/mesh-host/marker_test.go @@ -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 }) +} diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 2ee61ae..4842db6 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -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 diff --git a/internal/selfcompose/compose_while_passing_test.go b/internal/selfcompose/compose_while_passing_test.go new file mode 100644 index 0000000..234d1ef --- /dev/null +++ b/internal/selfcompose/compose_while_passing_test.go @@ -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") + } +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 6607fc6..1d359f3 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -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():