From c6df8fa20f11cdbdbb42a7dd6a6ad8a6a21c8e54 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 01:12:06 +0100 Subject: [PATCH] =?UTF-8?q?G=20=C2=A7683:=20a=20start=20composes=20the=20m?= =?UTF-8?q?odules=20whose=20kept=20want=20and=20entry=20it=20holds=20befor?= =?UTF-8?q?e=20its=20other=20inputs=20are=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The node's own record comes from the node-engine's statement of itself, taken in memory with no bus. Builds: design 47 G A-117, A-120, §683, A-95 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- cmd/mesh-host/nodeagent.go | 4 +- internal/link/session.go | 17 +++++ internal/selfcompose/entries_test.go | 11 +++- internal/selfcompose/selfcompose.go | 97 ++++++++++++++++++++++++++++ 4 files changed, 127 insertions(+), 2 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 0290447d..90c93e3d 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -702,6 +702,8 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus ag = agent.New(env) own.set(ag) current.set(ag) + // The node's own record is stated before the composer starts, from what the node knows of itself and with no bus (A-120). + sayFacts(run, nil) err = ag.Run(run) stopRun() cancel() @@ -1152,7 +1154,7 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f // composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer // to assignment candidates (selfcompose.Shadow) alike. func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { - env := selfcompose.Env{Node: node, + env := selfcompose.Env{Node: node, NodeRecord: session.OwnRecord, Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { if bucket == buckets.Assignments || bucket == buckets.ProposalCandidates { // The seat's states are made when their holder starts: a node-engine that starts first waits for the diff --git a/internal/link/session.go b/internal/link/session.go index 8bb427f5..74767dd5 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -183,6 +183,10 @@ func (q *Session) SayFacts(ctx context.Context, r Report) bool { bus := q.bus q.mu.Unlock() if bus == nil { + // With no link open the facts are still taken in memory, so the node's own record is known to what composes before the + // bus is read (design 47 G A-120); nothing is written. + r.Node = q.Membership.Node + q.noteApply(r) return false } r.Node = q.Membership.Node @@ -196,6 +200,19 @@ func (q *Session) SayFacts(ctx context.Context, r Report) bool { return true } +// OwnRecord is the record of this node the node-engine states of itself (tunnel key, sealing key, capabilities), from the facts it +// last took, held in memory (design 47 G A-120); false before any. +func (q *Session) OwnRecord() (nodeengine.NodeRecord, bool) { + q.init() + s := &q.haves + s.mu.Lock() + defer s.mu.Unlock() + if s.node == nil || s.node.Record == nil { + return nodeengine.NodeRecord{}, false + } + return *s.node.Record, true +} + // FactsAsked is the mesh asking this node to say its facts again: run on the link open now. func (q *Session) FactsAsked(ctx context.Context) { q.init() diff --git a/internal/selfcompose/entries_test.go b/internal/selfcompose/entries_test.go index 829659ae..f8b6196b 100644 --- a/internal/selfcompose/entries_test.go +++ b/internal/selfcompose/entries_test.go @@ -125,10 +125,19 @@ func composeFromEntries(t *testing.T) { mu.Unlock() env2 := b.env() env2.Entry, env2.ComposedState = EntryCall(ask), state + // The node's own record, stated by the node-engine of itself in memory (A-120). + env2.NodeRecord = func() (nodeengine.NodeRecord, bool) { + return nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}, true + } c2 := New(env2) - if _, err := c2.Watch(ctx, "n1.>"); err != nil { + out2, err := c2.Watch(ctx, "n1.>") + if err != nil { t.Fatal(err) } + // And composes the module it kept a want and an entry of before any input is read, nothing of the others (design 47 G §683, A-117). + if u := next(t, out2); u.Key != lemursKey || u.Deleted { + t.Errorf("the early pass handed %q, want the composition %s of lemurs", u.Key, lemursKey) + } c2.mu.Lock() kept, keptWant := c2.keptEnt[fp], c2.wants["lemurs"].Fp c2.mu.Unlock() diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index a87ec31a..633ff794 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -61,6 +61,9 @@ const ResolvedBucket = "mesh-config_resolved" // Env is what the composer needs of the node. type Env struct { Node string + // NodeRecord is the node's own record as the node-engine states it of itself, held in memory; the early pass of a start composes + // with it before the have bucket is read (design 47 G A-120). Nil, or false, has none. + NodeRecord func() (nodeengine.NodeRecord, bool) // EntryNotFound is told when the catalog half answered it holds no entry of a fingerprint (notFound), and when that is no longer so: // the entry answered, or nothing the node holds names it (design 47 G §614, A-117). Nil tells nobody. EntryNotFound func(fp string, notFound bool) @@ -241,6 +244,8 @@ type Composer struct { asked chan chan Pass // passing is set while a pass is made. passing atomic.Bool + // early is true while the early pass of a start runs (design 47 G §683, A-117). + early 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. @@ -611,6 +616,20 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er // input moves (issue 1126, 2026-10-28). var again <-chan time.Time var replies []chan Pass + // At a start, until its assignments, haves and resolved values are read, the node composes the modules whose kept want and + // kept entry it holds, so the bus module and the indexer start while both are down (design 47 G §683, A-95, A-117). + if c.standing() { + c.passing.Store(true) + updates := c.composeEarly(ctx) + c.passing.Store(false) + for _, u := range updates { + select { + case out <- u: + case <-ctx.Done(): + return + } + } + } for { // Drain whatever came, then compose once for all of it. select { @@ -975,6 +994,7 @@ func (c *Composer) takeValue(u agent.Update) { // compose composes the node once and answers the updates to hand: each composition whose digest differs from the one // handed, and a deletion for each handed one no want names any more. func (c *Composer) compose(ctx context.Context) []agent.Update { + early := c.early.Load() pass := Pass{} defer func() { pass.At = time.Now() @@ -1013,6 +1033,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { if module == agent.NodeModule { continue } + if early && !c.stands(module, w) { + continue + } fps[module] = w.Fp if m, ok := c.modules[module]; ok && m.Fp2 != w.Fp { behind = append(behind, module) @@ -1113,6 +1136,11 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { } else { c.paid() } + if early { + // Inputs not yet read may refuse it: nothing is marked, and the full pass follows when they are read. + c.env.Say("the node's early composition of the modules it kept is refused, and the full pass follows: " + why) + return nil + } pass.Refused = why // Said before the agent marks the modules: the marks read each have first, which a paused bus delays. c.sayOnce("the node cannot compose its modules, and keeps running what it runs: " + why) @@ -1232,6 +1260,9 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { } pubComps[key] = comp } + if early { + return out // what the early pass handed is composed from kept state alone: nothing is published from it + } c.pubAt = &published{scope: scope, fps: maps.Clone(fps), comps: pubComps, held: maps.Clone(held), placed: placedMods, ran: ranOf(running)} c.pubDue = true @@ -2431,3 +2462,69 @@ func (c *Composer) EngineModule() string { defer c.mu.Unlock() return c.engine } + +// stands says the want of a module stands for its placement in the early pass of a start: its want is a kept one the live read has +// not answered, and the entry of its build is kept (design 47 G A-117). +func (c *Composer) stands(module string, w meshwants.WantsByWantNodeModuleValue) bool { + if !c.byEntries() || !c.keptWants[module] { + return false + } + k, ok := c.keptEnt[w.Fp] + return ok && k.entry.Module != "" +} + +// standing says the composer holds a kept want and the kept entry of its build, so an early pass has something to compose. +func (c *Composer) standing() bool { + c.mu.Lock() + defer c.mu.Unlock() + for module, w := range c.wants { + if module != agent.NodeModule && c.stands(module, w) { + return true + } + } + return false +} + +// composeEarly is the pass of a start before the assignments, haves and resolved values are read: it composes only the modules whose +// kept want and kept entry it holds, a kept want standing for their placement, and leaves every other module as it runs; the full +// pass follows once the inputs are read (design 47 G §683, A-95, A-117). +func (c *Composer) composeEarly(ctx context.Context) []agent.Update { + node := c.env.Node + c.mu.Lock() + _, hadRecord := c.records[node] + if !hadRecord && c.env.NodeRecord != nil { + if rec, ok := c.env.NodeRecord(); ok { + c.records[node] = rec + } else { + hadRecord = true // nothing laid, nothing to take away + } + } + _, had := c.assignments[node] + if !had { + p := meshmoduleassigner.Assignments{Node: node} + for module, w := range c.wants { + if module != agent.NodeModule && c.stands(module, w) { + p.Modules = append(p.Modules, meshmoduleassigner.PlacedModule{Module: module}) + } + } + sort.Slice(p.Modules, func(i, j int) bool { return p.Modules[i].Module < p.Modules[j].Module }) + c.assignments[node] = p + } + c.mu.Unlock() + c.early.Store(true) + defer func() { + c.early.Store(false) + if !hadRecord { + c.mu.Lock() + delete(c.records, node) + c.mu.Unlock() + } + if !had { + c.mu.Lock() + delete(c.assignments, node) + c.mu.Unlock() + } + }() + c.env.Say("the node composes the modules it kept a want and an entry of, before its other inputs are read") + return c.compose(ctx) +}