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

A fill while the agent restarts is judged by the last view held and, refused, is owed and told again (issue 924)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 23:03:58 +01:00
parent 7add29cffe
commit c55672ce5e
4 changed files with 101 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/nodeagent.go
+34 -3
View File
@@ -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)
}
}
cmd/mesh-host/whatruns_test.go
+45
View File
@@ -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)
}
}
internal/filling/filling.go
+14 -1
View File
@@ -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
internal/filling/filling_test.go
+8
View File
@@ -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)