From c55672ce5ea0208d759e7729f125f77eabdf773c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 23:03:58 +0100 Subject: [PATCH] A fill while the agent restarts is judged by the last view held and, refused, is owed and told again (issue 924) Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- cmd/mesh-host/nodeagent.go | 37 +++++++++++++++++++++++--- cmd/mesh-host/whatruns_test.go | 45 ++++++++++++++++++++++++++++++++ internal/filling/filling.go | 15 ++++++++++- internal/filling/filling_test.go | 8 ++++++ 4 files changed, 101 insertions(+), 4 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 812c122..a1b4568 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -699,14 +699,36 @@ var reloadSettles = 10 * time.Second type agentNow struct { mu sync.Mutex ag *agent.Agent + // last is the view of what runs the previous agent held when it stopped, kept in memory: while the next agent has + // read no haves yet, what runs is this and never the retired kept declaration (a bus link that drops restarts the + // agent, and every fill in that window was refused; issue 924). + last *runsView +} + +// runsView is what an agent said runs, to be read after the agent stopped. +type runsView struct { + everything []compositions.Composed + holdsNode bool } func (c *agentNow) set(a *agent.Agent) { c.mu.Lock() + if a == nil && c.ag != nil { + if v := (&runsView{c.ag.Everything(), c.ag.HoldsNode()}); len(v.everything) > 0 || v.holdsNode { + c.last = v + } + } c.ag = a c.mu.Unlock() } +// lastView is the view the last agent held when it stopped, nil when none held any. +func (c *agentNow) lastView() *runsView { + c.mu.Lock() + defer c.mu.Unlock() + return c.last +} + func (c *agentNow) get() *agent.Agent { c.mu.Lock() defer c.mu.Unlock() @@ -890,11 +912,20 @@ func declaredNow(statePath string, signer ed25519.PublicKey) (*declaration.Decla func runsFromAgent(current *agentNow) func(string, ed25519.PublicKey) (*declaration.Declaration, error) { return func(statePath string, signer ed25519.PublicKey) (*declaration.Declaration, error) { kept := func() (*declaration.Declaration, error) { return keptDeclaration(statePath, signer) } - ag := current.get() - if ag == nil { + var view *runsView + if ag := current.get(); ag != nil { + view = &runsView{ag.Everything(), ag.HoldsNode()} + } + // An agent that has read no haves yet, or none: the last view held, never the retired kept declaration. + if view == nil || (len(view.everything) == 0 && !view.holdsNode) { + if last := current.lastView(); last != nil { + view = last + } + } + if view == nil { return kept() } - return whatRuns(ag.Everything(), ag.HoldsNode(), kept) + return whatRuns(view.everything, view.holdsNode, kept) } } diff --git a/cmd/mesh-host/whatruns_test.go b/cmd/mesh-host/whatruns_test.go index fbead93..8d0a787 100644 --- a/cmd/mesh-host/whatruns_test.go +++ b/cmd/mesh-host/whatruns_test.go @@ -10,6 +10,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" engcomp "github.com/novox/mesh-host/internal/compositions" + "os" "path/filepath" "sort" "strings" @@ -19,6 +20,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/declaration" "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/filling" "github.com/novox/mesh-host/internal/liveness" "github.com/novox/mesh-host/internal/store" ) @@ -237,3 +239,46 @@ func TestASharedTargetWithAnotherIDIsNotDropped(t *testing.T) { t.Errorf("a resource sharing only its target was dropped: %v", dropped) } } + +// Issue 924: the bus link drops, the agent restarts, and the filling is asked while it holds nothing yet. What runs is +// then the last view the node held, never the retired kept declaration; and a fill refused for want of any is owed and +// told once the agent holds the node again, so every file is written afterwards. +func TestAFillWhileTheAgentRestartsIsWrittenAfterwards(t *testing.T) { + keptRetired.Store(true) + defer keptRetired.Store(false) + node := engcomp.Composed{Node: "n1", Own: engcomp.NodeOwn{Provisioner: "mesh-node-provisioner"}} + files := []engcomp.Composed{{Node: "n1", Module: "db", Resources: []json.RawMessage{aFileOf("db", "a")}, Filled: []string{"/tmp/r6/db/a"}}} + view := &runsView{everything: append(files, node), holdsNode: true} + + var placed []string + current := &agentNow{} + e := &filling.Engine{ + Declared: func() (*declaration.Declaration, error) { return runsFromAgent(current)("/nowhere", nil) }, + UIDOf: func(string) (uint32, error) { return 4242, nil }, + Lock: func() (func(), error) { return func() {}, nil }, + Place: func(_ context.Context, _ *declaration.Declaration, path string, _ []byte) (string, string, error) { + placed = append(placed, path) + return "updated", "d", nil + }, + } + fill := func() filling.Answer { + return e.Write(context.Background(), 4242, filling.Request{Path: "/tmp/r6/db/a", Content: []byte("x")}) + } + socket := filepath.Join(t.TempDir(), "sock") + if err := os.WriteFile(socket, nil, 0o600); err != nil { + t.Fatal(err) + } + + // The node never held a view (a cold start): refused, and owed. + if a := fill(); a.Failed == "" && a.Refused == "" { + t.Fatalf("a fill with nothing read was accepted: %+v", a) + } + // The agent stopped holding the node: the last view it held answers, and the fill is written. + current.last = view + if told, err := e.Knock(socket); err != nil || told == "" { + t.Fatalf("the refused fill was not owed and told: %q, %v", told, err) + } + if a := fill(); a.Failed != "" || a.Refused != "" || len(placed) != 1 { + t.Fatalf("the fill during the restart was not written: %+v, placed %v", a, placed) + } +} diff --git a/internal/filling/filling.go b/internal/filling/filling.go index 4954d78..fd98e58 100644 --- a/internal/filling/filling.go +++ b/internal/filling/filling.go @@ -100,6 +100,13 @@ type owed struct { unnamed map[string]bool // refused: the declaration did not name it as filled failed map[string]bool // attempted, and the write or what reads it failed noAccount bool // refused: the declaration named no provisioner account + unread bool // refused: the declaration could not be read (a restart of the agent, a bus that is gone) +} + +func (o *owed) setUnread() { + o.mu.Lock() + o.unread = true + o.mu.Unlock() } func (o *owed) add(kind *map[string]bool, path string) { @@ -115,6 +122,7 @@ func (o *owed) add(kind *map[string]bool, path string) { func (e *Engine) Accepts(uid uint32) error { d, err := e.Declared() if err != nil { + e.owed.setUnread() return fmt.Errorf("this node's declaration cannot be read, so no write is accepted: %w", err) } if err := JudgeAccount(d, uid, e.UIDOf); err != nil { @@ -147,6 +155,7 @@ func (e *Engine) Write(ctx context.Context, uid uint32, req Request) Answer { defer unlock() d, err := e.Declared() if err != nil { + e.owed.setUnread() return Refusal(req.Path, "this node's declaration cannot be read, so nothing was written: "+err.Error()) } if err := JudgeAccount(d, uid, e.UIDOf); err != nil { @@ -184,7 +193,7 @@ func (e *Engine) Knock(socket string) (string, error) { } defer unlock() e.owed.mu.Lock() - nothing := len(e.owed.unnamed) == 0 && len(e.owed.failed) == 0 && !e.owed.noAccount + nothing := len(e.owed.unnamed) == 0 && len(e.owed.failed) == 0 && !e.owed.noAccount && !e.owed.unread e.owed.mu.Unlock() if nothing { return "", nil @@ -206,6 +215,10 @@ func (e *Engine) Knock(socket string) (string, error) { why = append(why, p+" (its write failed)") delete(e.owed.failed, p) } + if e.owed.unread { + why = append(why, "every file (the declaration can be read now)") + e.owed.unread = false + } if e.owed.noAccount && d.Provisioner != "" { why = append(why, "every file (the provisioner's account is named now)") e.owed.noAccount = false diff --git a/internal/filling/filling_test.go b/internal/filling/filling_test.go index c50fc34..b782836 100644 --- a/internal/filling/filling_test.go +++ b/internal/filling/filling_test.go @@ -118,7 +118,12 @@ func TestAnUndeclaredPathIsRefused(t *testing.T) { after string // the declaration applied afterwards knock string // what the knock names; "" for none twice bool // a second knock names nothing + // the declaration could not be read when the write came (the agent restarts): owed, and told once it can be + unreadable bool }{ + {name: "a write refused while the declaration could not be read", before: withAccount, unreadable: true, + account: provisionerUID, path: "/etc/m/m.conf", after: withAccount, + knock: "every file (the declaration can be read now)", twice: true}, {name: "an undeclared path the next declaration does not name either", before: withAccount, account: provisionerUID, path: "/etc/m/other.conf", after: withAccount}, {name: "an undeclared path the next declaration names", before: withAccount, account: provisionerUID, @@ -141,6 +146,9 @@ func TestAnUndeclaredPathIsRefused(t *testing.T) { return "", "", row.place } } + if row.unreadable { + e.Declared = func() (*declaration.Declaration, error) { return nil, errors.New("the agent restarts") } + } _ = e.Accepts(row.account) e.Write(context.Background(), row.account, Request{Path: row.path, Content: []byte("x")}) after := declared(t, row.after)