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 46a8fd7..812c122 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -596,6 +596,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus // No reader is no gate that lets everything through: the node acts on nothing new, as one that waits does. say(err.Error() + "; the node composes nothing new until it is started again and reads the list (novox/hq ADR 0636 §1)") } + // The marks of the haves run off the composition pass, each bounded (see marker). + marks := newMarker(run) hooks := composerHooks{ seatList: seats, seatListFailed: err != nil, @@ -621,13 +623,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..ed56149 --- /dev/null +++ b/internal/selfcompose/compose_while_passing_test.go @@ -0,0 +1,82 @@ +package selfcompose + +import ( + "context" + "strings" + "sync/atomic" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/buckets" + "github.com/novox/mesh-host/internal/buckets/bucketstest" + "github.com/novox/mesh-host/internal/compositions" +) + +// 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() []compositions.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) + } + place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}}) + have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) + nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true}) + b.putAssigned(t, place) + b.put(t, buckets.Have, "n1.node", have) + b.put(t, buckets.Wants, "want.n1.node", nodeWant) + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.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 080dca2..a0f89ad 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -191,6 +191,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. @@ -240,6 +242,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. @@ -331,6 +336,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(): @@ -539,7 +554,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():