Tell the provisioner's client to fill again once a settled apply may let through a write the filling refused (ace, 2026-10-25 21:17)
ace's filling refused mesh-build-go's files as not named as filled 32 s before its want arrived; after the apply that named them nothing asked again, and the runtime crash-looped on its missing broker credential. The filling now keeps what it refused or failed and, when the node agent says a module's apply or removal settled (agent.Env.Settled, called once what runs holds it), touches its write socket if the declaration now names such a path, names the provisioner's account, or a write had failed; the client watches that socket and fills every module again. Said in the journal each time.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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))
|
||||
|
||||
+12
-1
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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,73 @@ 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 after every apply (the node agent's afterApply): an event of the
|
||||
// engine's own, never a timer. 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.
|
||||
func (e *Engine) Knock(socket string) (string, error) {
|
||||
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 {
|
||||
|
||||
@@ -88,6 +88,76 @@ 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")
|
||||
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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnotherAccountIsRefused(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user