diff --git a/node-tools/cmd/node-tools/main.go b/node-tools/cmd/node-tools/main.go index 168250b..5796757 100644 --- a/node-tools/cmd/node-tools/main.go +++ b/node-tools/cmd/node-tools/main.go @@ -79,7 +79,10 @@ func main() { defer stopViews() gate := approval.New(approval.Deps{Node: conn.Node(), Runner: cred.Module, Catalog: source.ApprovalCatalog(definitionsOf(seatList)), Wait: source.WaitCatalog, Findings: source.Findings(), Logf: log.Printf, State: approvalsState{conn: conn, module: cred.Module}, - Publish: publishAsk(conn), Record: routerRecord(conn), Warranted: warranted(conn), Bounds: boundState{conn: conn, module: cred.Module}}) + Publish: publishAsk(conn), Record: routerRecord(conn), Warranted: warranted(conn), Bounds: boundState{conn: conn, module: cred.Module}, + Holder: approval.HoldersIn(func() (map[string][]byte, error) { return source.State(firstview.SeatHoldersRead) })}) + // A failed read or write of the state approvals is made again at the next membership change the runner hears (A-157). + conn.OnMembership(func(bus.Membership) { go gate.StateAgain() }) opts = append(opts, runtime.WithGate(gate)) // A run still unrecorded is asked again at each start and when the hand-act seat's holder is heard again (lead answer A-157). if stopBounds, err := gate.WatchBounds(); err != nil { diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index 2baddfb..9c42034 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -106,6 +106,9 @@ type world struct { runs int findings *findings bounds *bounds + // holders is the assigner's view of the seats' holders, by key; holdersErr says it is not answered yet. + holders map[string][]byte + holdersErr error // warranted is what mesh-hand-acts.warranted answers, and recorded each ask it answered for. warranted error recorded []string @@ -131,14 +134,16 @@ func (b *bounds) expire(key string) { func newWorld() *world { shell := entry("src2:1", []string{"node-login-shell.execute"}, map[string]any{"needs-approval": []any{"wipe"}}) w := &world{state: newMem(), records: map[string]seat.AsksByAskerIDValue{}, takes: true, findings: &findings{}, - bounds: &bounds{keys: map[string]time.Duration{}}, - now: time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC), + bounds: &bounds{keys: map[string]time.Duration{}}, + holders: map[string][]byte{"operator-channel": []byte(`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"messenger"}]}`)}, + now: time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC), cat: &fakeCatalog{revision: "7", byFp: map[string]catalogseat.CatalogEntry{"src2:1": shell}, newest: map[string]catalogseat.CatalogEntry{"shell": shell}, defs: map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", Verbs: []seatdef.Verb{{Name: "execute", NeedsApproval: true}}}}}} w.g = New(Deps{Node: "n1", Runner: "node-tools", Catalog: w.cat, State: w.state, Findings: w.findings, Bounds: w.bounds, - Go: func(f func()) { f() }, + Go: func(f func()) { f() }, + Holder: HoldersIn(func() (map[string][]byte, error) { return w.holders, w.holdersErr }), Warranted: func(asker, ask string, open func() (bool, error)) error { if still, _ := open(); !still { return nil @@ -495,6 +500,69 @@ func TestTheRefusalsAndAnswers(t *testing.T) { t.Fatalf("kept %+v", k) } }}, + {"§19: with no holder of operator-channel in the view, a marked call is refused asking nothing", func(t *testing.T, w *world) { + w.holders["operator-channel"] = []byte(`{"seat":"operator-channel","scope":"mesh","holders":[]}`) + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "no module holds operator-channel") || len(w.asked) != 0 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + w.holdersErr = errors.New("the first view has not been answered yet") + if _, err := w.do(call("wipe", `{}`)); err == nil || len(w.asked) != 0 { + t.Fatalf("an unread view: %v, asked %d", err, len(w.asked)) + } + }}, + {"§19: a publish that fails and a record that cannot be read refuse, running nothing", func(t *testing.T, w *world) { + pub := w.g.d.Publish + w.g.d.Publish = func(string, string, any, string) error { return errors.New("the bus refused it") } + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "the ask could not be raised") { + t.Fatalf("%v", err) + } + w.g.d.Publish = pub + w.state.vals = map[string]json.RawMessage{} + _, _ = w.do(call("wipe", `{}`)) + w.g.d.Record = func(string, string) (*stdio.StateEntry, error) { return nil, errors.New("timeout") } + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "the router's record of its ask") || w.runs != 0 { + t.Fatalf("%v, runs %d", err, w.runs) + } + }}, + {"§5, §19: a served entry that cannot be read refuses every marked call on that runtime", func(t *testing.T, w *world) { + c := call("wipe", `{}`) + c.Fingerprint = "src2:unknown" + if _, err := w.do(c); err == nil || !strings.Contains(err.Error(), "cannot be read") || len(w.asked) != 0 { + t.Fatalf("%v, asked %d", err, len(w.asked)) + } + }}, + {"§19: the marks unread, a call waits within 5 seconds and is refused naming mesh_first_views", func(t *testing.T, w *world) { + w.cat.ready = errors.New("the catalog half has no first view of main yet") + waited := time.Duration(0) + w.g.d.Wait = func(d time.Duration) error { waited = d; return w.cat.ready } + if _, err := w.do(call("tidy", `{}`)); err == nil || !strings.Contains(err.Error(), "mesh_first_views") || waited != RecordWithin || w.runs != 0 { + t.Fatalf("%v, waited %s", err, waited) + } + w.g.d.Wait = func(time.Duration) error { w.cat.ready = nil; return nil } + if _, err := w.do(call("tidy", `{}`)); err != nil || w.runs != 1 { + t.Fatalf("read within the wait: %v, runs %d", err, w.runs) + } + }}, + {"§19: the state failing refuses, raises approvals-unreadable after five minutes, and clears when read again", func(t *testing.T, w *world) { + stop, _ := w.g.WatchBounds() + defer stop() + w.state.fail = errors.New("the bus is unreachable") + if _, err := w.do(call("wipe", `{}`)); err == nil || !strings.Contains(err.Error(), "cannot be read") { + t.Fatalf("%v", err) + } + if w.bounds.keys["n1.unreadable.approvals"] != UnreadableAfter || len(w.findings.raised) != 0 { + t.Fatalf("bounds %v, findings %v", w.bounds.keys, w.findings.raised) + } + w.bounds.expire("n1.unreadable.approvals") + if w.findings.raised[keys.EscapeName("n1.approvals")] != KindApprovalsUnreadable { + t.Fatalf("findings %v", w.findings.raised) + } + w.state.fail = nil + w.g.StateAgain() // a membership change heard, no further call + if len(w.findings.raised) != 0 || len(w.bounds.keys) != 0 { + t.Fatalf("findings %v, bounds %v", w.findings.raised, w.bounds.keys) + } + }}, } { t.Run(tc.name, func(t *testing.T) { tc.run(t, newWorld()) }) } diff --git a/node-tools/internal/approval/ask.go b/node-tools/internal/approval/ask.go index a183f3f..43e1bc9 100644 --- a/node-tools/internal/approval/ask.go +++ b/node-tools/internal/approval/ask.go @@ -149,9 +149,11 @@ func (g *Gate) raise(c Call, act operatorchannelclient.Act, digest, markedSince return g.refused(c, digest, "the ask could not be raised: "+err.Error()) } if err := g.keepCall(id, act, c.Args); err != nil { + g.stateFailed(err) return g.refused(c, digest, "the ask was raised and its call could not be kept: "+err.Error()) } if _, err := g.d.State.Put(g.actKey(digest), id); err != nil { + g.stateFailed(err) return g.refused(c, digest, "the ask was raised and its act could not be kept: "+err.Error()) } return g.firstAnswer(c, id, digest, markedSince) diff --git a/node-tools/internal/approval/closed.go b/node-tools/internal/approval/closed.go new file mode 100644 index 0000000..96219a4 --- /dev/null +++ b/node-tools/internal/approval/closed.go @@ -0,0 +1,119 @@ +package approval + +import ( + "encoding/json" + "fmt" + "time" + + assignerseat "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" +) + +// AskedOn is the seat an ask is raised on, whose holder must be there for a marked call to be asked (§19). +const AskedOn = "operator-channel" + +// UnreadableAfter is how long the state approvals fails before the runner's finding says so (§19). +const UnreadableAfter = 5 * time.Minute + +// unreadableKey is the bound key whose expiry raises `approvals-unreadable` (lead answer A-157). +func (g *Gate) unreadableKey() string { return g.d.Node + ".unreadable." + StateName } + +// held says why no marked call can be asked now, or "" (§19): no module holds operator-channel, as the assigner's view +// of the seat's holders says, or that view is not read. +func (g *Gate) held() string { + if g.d.Holder == nil { + return "the tool runner reads no view of the seats' holders, so it cannot tell that a module holds " + AskedOn + } + ok, err := g.d.Holder(AskedOn) + switch { + case err != nil: + return fmt.Sprintf("whether a module holds %s cannot be read: %v", AskedOn, err) + case !ok: + return "no module holds " + AskedOn + ", so nobody can be asked" + } + return "" +} + +// stateFailed notes that the state approvals could not be read or written: no marked call runs on this machine until it +// can, the failed read or write is made again at the next marked call and at the next membership change (A-157), and +// five minutes on the runner's finding says it (§19). +func (g *Gate) stateFailed(err error) { + g.mu.Lock() + first := g.failingSince.IsZero() + if first { + g.failingSince = g.d.Now() + } + g.mu.Unlock() + if first { + g.d.Logf("[mesh-tools] the state %s on %s cannot be read or written (%v): no marked call runs here until it can", StateName, g.d.Node, err) + if g.d.Bounds != nil { + if e := g.d.Bounds.CreateTTL(g.unreadableKey(), map[string]string{"since": g.d.Now().UTC().Format(time.RFC3339)}, UnreadableAfter); e != nil { + g.d.Logf("[mesh-tools] the bound of the state %s's failure is not kept: %v", StateName, e) + } + } + } +} + +// stateRead notes that the state approvals was read or written: a failure noted before ends, its finding cleared. +func (g *Gate) stateRead() { + g.mu.Lock() + was := !g.failingSince.IsZero() + g.failingSince = time.Time{} + g.mu.Unlock() + if !was { + return + } + if g.d.Bounds != nil { + _ = g.d.Bounds.Delete(g.unreadableKey()) + } + g.finding(StateName, KindApprovalsUnreadable, false, "the state "+StateName+" is read again", "") +} + +// failing says the state approvals is failing now. +func (g *Gate) failing() bool { + g.mu.Lock() + defer g.mu.Unlock() + return !g.failingSince.IsZero() +} + +// StateAgain makes the failed read of the state approvals again: at the next membership change the runner hears (A-157). +func (g *Gate) StateAgain() { + if g.d.State == nil || !g.failing() { + return + } + if _, err := g.d.State.Keys(); err != nil { + return + } + g.stateRead() +} + +// unreadableExpired is the bound key of a failing state expiring: the finding raised while the state still fails. +func (g *Gate) unreadableExpired() { + g.mu.Lock() + since := g.failingSince + g.mu.Unlock() + if since.IsZero() { + return + } + g.finding(StateName, KindApprovalsUnreadable, true, fmt.Sprintf("the state %s on %s cannot be read or written since %s: no marked call "+ + "runs here until it can", StateName, g.d.Node, since.UTC().Format(time.RFC3339)), "approvals unreadable on "+g.d.Node) +} + +// HoldersIn is Holder over the assigner's view of the seats' holders as the runner holds it (§19): view answers the view's +// values by key, an error while its first view is not answered; a seat is held when its key names a holder. +func HoldersIn(view func() (map[string][]byte, error)) func(seat string) (bool, error) { + return func(seat string) (bool, error) { + values, err := view() + if err != nil { + return false, err + } + raw, ok := values[assignerseat.SeatHoldersBySeatKey(seat)] + if !ok { + return false, nil + } + var h assignerseat.SeatHolders + if err := json.Unmarshal(raw, &h); err != nil { + return false, fmt.Errorf("the holders of %s cannot be read: %w", seat, err) + } + return len(h.Holders) > 0, nil + } +} diff --git a/node-tools/internal/approval/gate.go b/node-tools/internal/approval/gate.go index 5381ec9..419197c 100644 --- a/node-tools/internal/approval/gate.go +++ b/node-tools/internal/approval/gate.go @@ -53,6 +53,8 @@ type Deps struct { Warranted func(asker, ask string, open func() (bool, error)) error // Bounds is the runner's key-ttl state bound (A-157, A-158); nil keeps no bound. Bounds Bounds + // Holder says whether a module holds a seat, from the assigner's view of the seats' holders the runner follows (§19). + Holder func(seat string) (bool, error) // Go runs what goes on beside the call (a run's recording); nil is a goroutine. Go func(func()) } @@ -69,6 +71,8 @@ type Gate struct { unmarkedAt map[string]bool // inFlight is each ask whose recording runs now. inFlight map[string]bool + // failingSince is when the state approvals began to fail; zero while it is read. + failingSince time.Time } // New is a gate over d. diff --git a/node-tools/internal/approval/record.go b/node-tools/internal/approval/record.go index a01a038..20af7ab 100644 --- a/node-tools/internal/approval/record.go +++ b/node-tools/internal/approval/record.go @@ -200,6 +200,10 @@ func (g *Gate) WatchBounds() (func(), error) { return func() {}, nil } return g.d.Bounds.Watch(g.d.Node+".**", func(key string) { + if key == g.unreadableKey() { + g.unreadableExpired() + return + } rest := strings.TrimPrefix(key, g.d.Node+".") kind, id, _ := strings.Cut(rest, ".") if open, err := g.unrecorded(id); err != nil || !open { diff --git a/node-tools/internal/approval/run.go b/node-tools/internal/approval/run.go index c828e6f..b0f525c 100644 --- a/node-tools/internal/approval/run.go +++ b/node-tools/internal/approval/run.go @@ -38,6 +38,9 @@ func (g *Gate) Run(c Call, run func() (any, error)) (any, error) { if why := refusedUnasked(c.Args); why != "" { return nil, g.refused(c, "", why) } + if why := g.held(); why != "" { + return nil, g.refused(c, "", why) + } act, err := ActOf(c, g.d.Node) if err != nil { return nil, g.refused(c, "", err.Error()) @@ -48,8 +51,10 @@ func (g *Gate) Run(c Call, run func() (any, error)) (any, error) { } named, err := g.d.State.Get(g.actKey(digest)) if err != nil { + g.stateFailed(err) return nil, g.refused(c, digest, "its state "+StateName+" cannot be read: "+err.Error()) } + g.stateRead() if named == nil { return nil, g.raise(c, act, digest, since) } @@ -163,6 +168,7 @@ func (g *Gate) judge(c Call, act operatorchannelclient.Act, digest, id, since st case h == operatorchannelclient.Refused: return nil, g.refused(c, digest, "its warrant does not hold: "+errText(err)) case h == "" && err != nil: + g.stateFailed(err) return nil, g.refused(c, digest, "its state "+StateName+" cannot be written: "+err.Error()) case err != nil: // the act about to run is not the one the option bound: nothing runs, the warrant stays used return nil, g.refused(c, digest, err.Error())