Merge pull request 'The node-engine wakes the provisioner when an apply lets a file it refused be written (issue 924)' (#142) from fix/provisioner-retries-on-declaration into main

cmd/mesh-host/filling.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"crypto/ed25519"
 	"errors"
+	"fmt"
 	"path/filepath"
+	"sync/atomic"
 
 	"git.novox.be/novox/mesh-sdk/go/declaration"
 	"github.com/novox/mesh-host/internal/apply"
@@ -16,11 +18,34 @@ import (
 // serveFilling writes filled files on the provisioner client's request for as long as this engine serves (novox/hq
 // ADR 0349 §4). A socket that cannot be served is said once, and this engine goes on applying.
 func serveFilling(ctx context.Context, opts options, signer ed25519.PublicKey, say func(string)) {
-	if err := filling.Serve(ctx, filling.Socket, fillingEngine(opts, signer, say), say); err != nil {
+	e := fillingEngine(opts, signer, say)
+	fillingNow.Store(e)
+	defer fillingNow.CompareAndSwap(e, nil)
+	if err := filling.Serve(ctx, filling.Socket, e, say); err != nil {
 		say("filled files are not written on request on this machine: " + err.Error())
 	}
 }
 
+// fillingNow is the filling this engine serves, for knockFilling; nil while none is served.
+var fillingNow atomic.Pointer[filling.Engine]
+
+// knockFilling tells the provisioner's client to fill again what the filling refused or failed and the declaration
+// applied now may let through (filling.Engine.Knock), once a module's apply or removal settled (agent.Env.Settled).
+// What it told, or why it could not, is said.
+func knockFilling(module string, say func(string)) {
+	e := fillingNow.Load()
+	if e == nil {
+		return
+	}
+	told, err := e.Knock(filling.Socket)
+	switch {
+	case err != nil:
+		say(fmt.Sprintf("after %s settled: %v", module, err))
+	case told != "":
+		say(fmt.Sprintf("after %s settled, the provisioner's client is told to fill again: %s", module, told))
+	}
+}
+
 // fillingEngine is the write of a filled file as this engine does it: the declaration it keeps, verified, read under
 // the apply's lock — the same two locks every apply takes, the process's and the one beside the state — and the file
 // written through the apply (apply.WriteFilled), its outcome recorded in the node's state as an apply's is.

cmd/mesh-host/nodeagent.go

@@ -413,6 +413,9 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess
 			},
 			Trusted: trustedKeys(opts, signer),
 			Backup:  backupNode(session),
+			// What the filling refused before this apply may be named now: the provisioner's client is told to fill
+			// again (ace, 2026-10-25 21:17).
+			Settled: func(module string) { knockFilling(module, say) },
 			Say:     say,
 			Apply: func(ctx context.Context, c delivery.Composed, others []delivery.Composed) agent.Result {
 				actx, done := applyContext(session.WithLink(ctx))

internal/agent/agent.go

@@ -104,7 +104,12 @@ type Env struct {
 	// Revisions are a composition key's revisions the bucket keeps, oldest first; nil reads none. The composition a want
 	// names is taken from them when a newer one replaced it under the same key (ADR 0539 clause 1).
 	Revisions func(ctx context.Context, key string) ([]Update, error)
-	Say       func(string)
+	// Settled is told a module's apply or removal that succeeded, once what runs holds it (Everything), outside the
+	// agent's lock: the declaration the node applies now names what that composition names, its filled files among
+	// them. The node-engine's filling tells the provisioner's client then to fill again what it refused (mesh-host
+	// internal/filling Knock; ace, 2026-10-25 21:17). Nil tells nothing.
+	Settled func(module string)
+	Say     func(string)
 }
 
 // Agent follows the wants.
@@ -741,6 +746,9 @@ func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNo
 	a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning}
 	a.applied[module] = comp
 	a.mu.Unlock()
+	if a.env.Settled != nil {
+		a.env.Settled(module)
+	}
 	a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail})
 }
 
@@ -769,6 +777,9 @@ func (a *Agent) remove(ctx context.Context, module string, w wantNow) {
 	delete(a.attempt, module)
 	a.running[module] = delivery.Have{Phase: delivery.PhaseRemoved}
 	a.mu.Unlock()
+	if a.env.Settled != nil {
+		a.env.Settled(module)
+	}
 	a.env.Haves.Removed(ctx, module)
 }
 

internal/agent/agent_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ed25519"
 	"encoding/json"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -112,7 +113,11 @@ type bed struct {
 	kept     map[string][]json.RawMessage
 	others   [][]delivery.Composed
 	said     []string
-	env      Env
+	// settled is each module Settled was told, and settledRan whether what runs held it then.
+	settled    []string
+	settledRan []bool
+	ag         *Agent // the agent run() runs
+	env        Env
 }
 
 func newBed(t *testing.T) *bed {
@@ -147,6 +152,18 @@ func newBed(t *testing.T) *bed {
 			}
 			return Backup{Done: true}
 		},
+		Settled: func(m string) {
+			ran := false
+			if b.ag != nil {
+				for _, c := range b.ag.Everything() {
+					ran = ran || c.Module == m
+				}
+			}
+			b.mu.Lock()
+			defer b.mu.Unlock()
+			b.settled = append(b.settled, m)
+			b.settledRan = append(b.settledRan, ran)
+		},
 		Kept:    func() map[string][]json.RawMessage { return b.kept },
 		Trusted: func() []ed25519.PublicKey { return []ed25519.PublicKey{controllerPub, operatorPub} },
 		Say: func(s string) {
@@ -162,7 +179,8 @@ func newBed(t *testing.T) *bed {
 func (b *bed) run() (stop func()) {
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan struct{})
-	go func() { defer close(done); _ = New(b.env).Run(ctx) }()
+	b.ag = New(b.env)
+	go func() { defer close(done); _ = b.ag.Run(ctx) }()
 	return func() { cancel(); <-done }
 }
 
@@ -259,6 +277,13 @@ func TestAWantAndItsCompositionApplyTheModuleAndTheHaveNamesBoth(t *testing.T) {
 		h := b.haves.last("web")
 		return h.FP == fp && h.Composed == dig && h.Phase == string(delivery.PhaseRunning)
 	})
+	// Settled is told once what runs holds the module, so a declaration read then names its files (the filling's Knock;
+	// ace, 2026-10-25 21:17).
+	eventually(t, "Settled was not told web with web in what runs", func() bool {
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		return slices.Equal(b.settled, []string{"web"}) && b.settledRan[0]
+	})
 
 	// A want naming a digest the composition at that fingerprint does not have: nothing is applied.
 	fp2 := fpOf("b")

internal/filling/filling.go

@@ -29,6 +29,7 @@ import (
 	"path/filepath"
 	"sort"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -87,6 +88,28 @@ type Engine struct {
 	// change. Called holding the lock, with the declaration read under it.
 	Place func(ctx context.Context, d *declaration.Declaration, path string, content []byte) (action, digest string,
 		err error)
+
+	// owed is what this engine refused or failed to write since it last told the client to fill again (Knock).
+	owed owed
+}
+
+// owed is the writes a later declaration may let through: a path refused as not named as filled, a write refused
+// while the declaration named no provisioner account, and a path whose write failed. The client is told to fill
+// again only once a declaration may end one (Knock): it has no other way to learn that the declaration changed.
+type owed struct {
+	mu        sync.Mutex
+	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
+}
+
+func (o *owed) add(kind *map[string]bool, path string) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	if *kind == nil {
+		*kind = map[string]bool{}
+	}
+	(*kind)[path] = true
 }
 
 // Accepts reads the declaration and judges the account.
@@ -95,7 +118,21 @@ func (e *Engine) Accepts(uid uint32) error {
 	if err != nil {
 		return fmt.Errorf("this node's declaration cannot be read, so no write is accepted: %w", err)
 	}
-	return JudgeAccount(d, uid, e.UIDOf)
+	if err := JudgeAccount(d, uid, e.UIDOf); err != nil {
+		e.oweAccount(d)
+		return err
+	}
+	return nil
+}
+
+// oweAccount remembers a refusal a declaration naming the provisioner's account may end.
+func (e *Engine) oweAccount(d *declaration.Declaration) {
+	if d.Provisioner != "" {
+		return // another account than the client's: no declaration lets it through
+	}
+	e.owed.mu.Lock()
+	e.owed.noAccount = true
+	e.owed.mu.Unlock()
 }
 
 // Write takes the apply's lock, reads the declaration under it, judges the account and the path again — the
@@ -114,18 +151,87 @@ func (e *Engine) Write(ctx context.Context, uid uint32, req Request) Answer {
 		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 {
+		e.oweAccount(d)
 		return Refusal(req.Path, err.Error())
 	}
 	if err := JudgePath(d, req.Path); err != nil {
+		e.owed.add(&e.owed.unnamed, req.Path)
 		return Refusal(req.Path, err.Error())
 	}
 	action, digest, err := e.Place(ctx, d, req.Path, req.Content)
 	if err != nil {
+		e.owed.add(&e.owed.failed, req.Path)
 		return Answer{Path: req.Path, Action: action, Digest: digest, Failed: err.Error()}
 	}
 	return Answer{Path: req.Path, Action: action, Digest: digest}
 }
 
+// Knock tells the provisioner's client to fill again when the declaration this node applies now may let through a
+// write this engine refused or failed (issue 1873; ace, 2026-10-25 21:17: mesh-build-go's files were refused as not
+// named as filled 32 s before its want arrived, and nothing filled them after the apply that named them): a path
+// refused as not named that the declaration now names, a write refused while no provisioner account was named once
+// one is, and any path whose write failed. Called once a module's apply or removal settled (the node agent's
+// Settled, after what runs holds it): an event of the engine's own, never a timer. It holds the apply's lock, as a
+// write does, so a write judged by the declaration before the apply settled has recorded its refusal before Knock
+// reads what is owed, and one judged after is judged by the settled declaration. The client is told by touching the
+// write socket, whose attributes it watches (mesh-provisioner-client socketwatch.go, IN_ATTRIB), and then fills every
+// module it holds again. A refusal the declaration still does not end is kept for the next apply. It answers what it
+// told the client of, for the journal; nothing when it told nothing. With nothing owed it reads no declaration.
+func (e *Engine) Knock(socket string) (string, error) {
+	unlock, err := e.Lock()
+	if err != nil {
+		return "", fmt.Errorf("the apply's lock could not be taken, so the provisioner's client is not told to fill "+
+			"again: %w", err)
+	}
+	defer unlock()
+	e.owed.mu.Lock()
+	nothing := len(e.owed.unnamed) == 0 && len(e.owed.failed) == 0 && !e.owed.noAccount
+	e.owed.mu.Unlock()
+	if nothing {
+		return "", nil
+	}
+	d, err := e.Declared()
+	if err != nil {
+		return "", fmt.Errorf("this node's declaration cannot be read, so the provisioner's client is not told to "+
+			"fill again: %w", err)
+	}
+	e.owed.mu.Lock()
+	var why []string
+	for _, p := range sortedSet(e.owed.unnamed) {
+		if d.Fills(p) {
+			why = append(why, p+" (now named as filled)")
+			delete(e.owed.unnamed, p)
+		}
+	}
+	for _, p := range sortedSet(e.owed.failed) {
+		why = append(why, p+" (its write failed)")
+		delete(e.owed.failed, p)
+	}
+	if e.owed.noAccount && d.Provisioner != "" {
+		why = append(why, "every file (the provisioner's account is named now)")
+		e.owed.noAccount = false
+	}
+	e.owed.mu.Unlock()
+	if len(why) == 0 {
+		return "", nil
+	}
+	now := time.Now()
+	if err := os.Chtimes(socket, now, now); err != nil {
+		return "", fmt.Errorf("the provisioner's client could not be told to fill %s again: %w",
+			strings.Join(why, ", "), err)
+	}
+	return strings.Join(why, ", "), nil
+}
+
+func sortedSet(m map[string]bool) []string {
+	out := make([]string, 0, len(m))
+	for k := range m {
+		out = append(out, k)
+	}
+	sort.Strings(out)
+	return out
+}
+
 // JudgeAccount refuses every account but the one the declaration names as the provisioner's client's, by its user
 // id here — and every account while the declaration names none.
 func JudgeAccount(d *declaration.Declaration, uid uint32, uidOf func(string) (uint32, error)) error {

internal/filling/filling_test.go

@@ -88,6 +88,138 @@ func TestAnUndeclaredPathIsRefused(t *testing.T) {
 	if len(e.placed) != 0 {
 		t.Errorf("placed %v", e.placed)
 	}
+
+	// A refusal a later declaration ends is followed by a knock on the socket, which the provisioner's client watches,
+	// once a declaration applied after it ends it, and only then (ace, 2026-10-25 21:17: mesh-build-go's files refused
+	// 32 s before its want, and never filled after the apply that named them).
+	socket := filepath.Join(t.TempDir(), "fill.sock")
+	if err := os.WriteFile(socket, nil, 0o666); err != nil {
+		t.Fatal(err)
+	}
+	long := time.Now().Add(-time.Hour)
+	knocked := func() bool {
+		info, err := os.Stat(socket)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return info.ModTime().After(long.Add(time.Minute))
+	}
+	namesOther := strings.NewReplacer(`"filled_files":["/etc/m/m.conf"]`,
+		`"filled_files":["/etc/m/m.conf","/etc/m/other.conf"]`, `"content":"y"}`,
+		`"content":"y","owner-at-runtime":"owned"}`).Replace(withAccount)
+	failing := errors.New("is a directory")
+	var mu2 sync.Mutex // guards the race row's count of declaration reads
+	for _, row := range []struct {
+		name    string
+		before  string // the declaration the write is judged by
+		account uint32
+		path    string
+		place   error
+		after   string // the declaration applied afterwards
+		knock   string // what the knock names; "" for none
+		twice   bool   // a second knock names nothing
+	}{
+		{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,
+			path: "/etc/m/other.conf", after: namesOther, knock: "/etc/m/other.conf (now named as filled)", twice: true},
+		{name: "a write refused while no account was named, then named", before: withoutAccount,
+			account: provisionerUID, path: "/etc/m/m.conf", after: withAccount,
+			knock: "every file (the provisioner's account is named now)", twice: true},
+		{name: "a write that failed", before: withAccount, account: provisionerUID, path: "/etc/m/m.conf",
+			place: failing, after: withAccount, knock: "/etc/m/m.conf (its write failed)", twice: true},
+		{name: "another account, refused whatever is declared", before: withAccount, account: 1000,
+			path: "/etc/m/m.conf", after: withAccount},
+	} {
+		t.Run(row.name, func(t *testing.T) {
+			if err := os.Chtimes(socket, long, long); err != nil {
+				t.Fatal(err)
+			}
+			e := newEngine(t, row.before, &sync.Mutex{})
+			if row.place != nil {
+				e.Place = func(context.Context, *declaration.Declaration, string, []byte) (string, string, error) {
+					return "", "", row.place
+				}
+			}
+			_ = e.Accepts(row.account)
+			e.Write(context.Background(), row.account, Request{Path: row.path, Content: []byte("x")})
+			after := declared(t, row.after)
+			e.Declared = func() (*declaration.Declaration, error) { return after, nil }
+			told, err := e.Knock(socket)
+			if err != nil || told != row.knock || knocked() != (row.knock != "") {
+				t.Fatalf("knock told %q (%v), socket touched %v; want %q", told, err, knocked(), row.knock)
+			}
+			if row.twice {
+				if err := os.Chtimes(socket, long, long); err != nil {
+					t.Fatal(err)
+				}
+				if told, _ := e.Knock(socket); told != "" || knocked() {
+					t.Fatalf("a second knock told %q: what was told is no longer owed", told)
+				}
+			}
+		})
+	}
+
+	// The race (review of this fix): a write judged by the declaration before the apply settled, its refusal not yet
+	// recorded when Knock runs. Knock waits for the apply's lock the write holds, so it sees that refusal and knocks.
+	t.Run("a write still judging when the apply settles", func(t *testing.T) {
+		if err := os.Chtimes(socket, long, long); err != nil {
+			t.Fatal(err)
+		}
+		before, after := declared(t, withAccount), declared(t, namesOther)
+		judging, release := make(chan struct{}), make(chan struct{})
+		var mu sync.Mutex
+		knockLocking := make(chan struct{}, 1)
+		calls := 0
+		e := newEngine(t, withAccount, &mu)
+		e.Declared = func() (*declaration.Declaration, error) {
+			mu2.Lock()
+			calls++
+			first := calls == 1
+			mu2.Unlock()
+			if first {
+				close(judging)
+				<-release
+				return before, nil
+			}
+			return after, nil
+		}
+		e.Lock = func() (func(), error) {
+			mu2.Lock()
+			second := calls > 0
+			mu2.Unlock()
+			if second {
+				knockLocking <- struct{}{}
+			}
+			mu.Lock()
+			return mu.Unlock, nil
+		}
+		wrote := make(chan Answer)
+		go func() { wrote <- e.Write(context.Background(), provisionerUID, Request{Path: "/etc/m/other.conf"}) }()
+		<-judging
+		type result struct {
+			told string
+			err  error
+		}
+		knocked2 := make(chan result, 1)
+		go func() { told, err := e.Knock(socket); knocked2 <- result{told, err} }()
+		// Released once Knock waits for the lock, or once it has answered without taking it.
+		var r result
+		select {
+		case <-knockLocking:
+			close(release)
+			r = <-knocked2
+		case r = <-knocked2:
+			close(release)
+		}
+		if a := <-wrote; a.Refused == "" {
+			t.Fatalf("the write was not refused: %+v", a)
+		}
+		if r.err != nil || r.told != "/etc/m/other.conf (now named as filled)" || !knocked() {
+			t.Fatalf("knock told %q (%v), socket touched %v: the refusal still being judged was missed", r.told, r.err,
+				knocked())
+		}
+	})
 }
 
 func TestAnotherAccountIsRefused(t *testing.T) {