novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mesh-grants: a first view's condition the keeper did not take is said again at the next pass

Browse Source 
The raise and clear of first-view-unanswered dropped a failed publish (_ = say). The binding comment
named the bus module's own stream and 30 days; the part hears the seat's events on the shared stream, 7 days.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
jschoubben
2026-11-04 17:08:23 +01:00
parent 4dbb6b2aec
commit 4f7ceb61ea
4 changed files with 81 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-grants/grants_test.go
+16 -4
View File
@@ -2351,7 +2351,17 @@ func TestGrants(t *testing.T) {
}
var said []oc.Condition
var sayMu sync.Mutex
say := func(c oc.Condition) error { sayMu.Lock(); said = append(said, c); sayMu.Unlock(); return nil }
// The condition keeper does not take the first raise: it is said again until it lands, never dropped.
var refusedOnce atomic.Bool
say := func(c oc.Condition) error {
if c.State == occ.StateRaised && strings.HasPrefix(c.Key, grants.FirstViewUnanswered) && refusedOnce.CompareAndSwap(false, true) {
return errors.New("no responders")
}
sayMu.Lock()
said = append(said, c)
sayMu.Unlock()
return nil
}
fv := grants.NewViews(owner, say, func() {}, nil)
fv.Rest = b
var minutes atomic.Int64
@@ -2379,12 +2389,14 @@ func TestGrants(t *testing.T) {
}
return last == state
}
waitFor(t, func() bool {
return raised(grants.FirstViewUnanswered+".mesh-delivery-indexer.verdicts", occ.StateRaised)
})
waitFor(t, refusedOnce.Load)
p := partOn(b)
p.Read, p.Views, p.Say = fv, fv, say
before := p.Compose()
// The raise the condition keeper did not take was said again by the pass.
if !raised(grants.FirstViewUnanswered+".mesh-delivery-indexer.verdicts", occ.StateRaised) {
t.Fatalf("the refused raise was not said again at the pass: %+v", said)
}
if !slices.ContainsFunc(before.UserList.Gaps, func(g grants.Gap) bool { return g.Input == grants.ReadPlacement && g.Failed }) {
t.Fatalf("placement answered empty at revision 0 was not a failed read: %+v", before.UserList.Gaps)
}
modules/mesh-grants/internal/grants/bindings.go
+6 -7
View File
@@ -14,14 +14,13 @@ import (
// **The binding travels on the bus module's events** (novox/hq ADR 0595 §1, §7; design 25 § 12.2; DLV-G1). The bus
// module (the holder of the seat mesh-broker) says `created` when it created one entry of a module's bus state,
// `recreated` when it rebuilt one, and `dropped` when it removed one; each carries the entry's names on the bus. The
// grants part hears them on its durable consumer of the bus module's own stream (its manifest's `consumes`), writes the
// binding into the module's membership on every machine the module is assigned, rewrites it on `recreated` and removes
// it on `dropped`. What it heard is kept in its own state memberships, one key per module (BoundKey), so a binding
// outlives the 30 days its event is kept and reaches a machine the module is assigned to later. A `refused` create
// hands no binding: only these three events are read.
// grants part hears them as the seat's events, `mesh-broker.<event>` on the shared events stream (its manifest's
// `consumes`), on its durable consumer; writes the binding into the module's membership on every machine the module is
// assigned, rewrites it on `recreated` and removes it on `dropped`. What it heard is kept in its own state memberships,
// one key per module (BoundKey), so a binding outlives the 7 days the shared events stream keeps its event and reaches a
// machine the module is assigned to later. A `refused` create hands no binding: only these three events are read.
// The bus module's events the part hears, by their local names (ADR 0595 §6; ADR 0620 §2: the holder's own events,
// never a seat's subject).
// The bus module's events the part hears, by their local names on the seat mesh-broker (ADR 0595 §6).
const (
BrokerCreated = "created"
BrokerRecreated = "recreated"
modules/mesh-grants/internal/grants/firstviews.go
+57 -6
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"maps"
"slices"
"sort"
"strings"
@@ -76,6 +77,8 @@ type Views struct {
// Rest reads every other state, and Watching watches it (nil: watched by nobody, as in a tool).
Rest Reader
Watching Watcher
said *untilSaid
}
// NewViews makes the five followers: ask calls an owner's tool (through the part's own account), say says a condition,
@@ -83,20 +86,18 @@ type Views struct {
func NewViews(ask func(tool string) func(context.Context, string) (views.Answer, error), say func(oc.Condition) error,
changed func(), logf func(string, ...any)) *Views {
v := &Views{Followers: map[string]*views.Follower{}}
sayer := &untilSaid{say: say, logf: logf, unsaid: map[string]oc.Condition{}}
v.said = sayer
for _, fv := range FirstViews {
fv := fv
f := &views.Follower{Owner: fv.Owner, Tool: fv.Tool, State: fv.State, Ask: ask(fv.Tool), Logf: logf,
Changed: func(string, json.RawMessage, bool, time.Time) { changed() }}
f.Raise = func(s views.Status) {
if say != nil {
_ = say(unansweredCondition(fv, s, true))
}
sayer.Say(unansweredCondition(fv, s, true))
changed()
}
f.Clear = func(s views.Status) {
if say != nil {
_ = say(unansweredCondition(fv, s, false))
}
sayer.Say(unansweredCondition(fv, s, false))
changed()
}
v.Followers[fv.Read] = f
@@ -104,6 +105,56 @@ func NewViews(ask func(tool string) func(context.Context, string) (views.Answer,
return v
}
// untilSaid says a first view's condition; one the condition keeper did not take is logged and kept, and said again at
// the part's next pass (SayUnsaid), until it lands or a newer word on the same key replaces it: a raise or a clear is
// never dropped on a failed publish, and nothing waits on a timer for it.
type untilSaid struct {
say func(oc.Condition) error
logf func(string, ...any)
mu sync.Mutex
unsaid map[string]oc.Condition
}
// Say says c; when it fails it is kept to be said again.
func (u *untilSaid) Say(c oc.Condition) {
if u.say == nil {
return
}
err := u.say(c)
u.mu.Lock()
defer u.mu.Unlock()
if err == nil {
delete(u.unsaid, c.Key)
return
}
u.unsaid[c.Key] = c
if u.logf != nil {
u.logf("the condition %s (%s) could not be said: %v; said again at the next pass", c.Key, c.State, err)
}
}
// SayUnsaid says again every condition not taken yet: the part's pass calls it.
func (v *Views) SayUnsaid() {
if v == nil || v.said == nil {
return
}
u := v.said
u.mu.Lock()
pending := make([]oc.Condition, 0, len(u.unsaid))
for _, k := range slices.Sorted(maps.Keys(u.unsaid)) {
pending = append(pending, u.unsaid[k])
}
u.mu.Unlock()
for _, c := range pending {
u.mu.Lock()
current, still := u.unsaid[c.Key]
u.mu.Unlock()
if still && current.State == c.State {
u.Say(c)
}
}
}
// Start asks each first view on a goroutine of its own, until it is answered or ctx ends; each answer wakes a pass.
func (v *Views) Start(ctx context.Context, changed func()) {
for _, f := range v.Followers {
modules/mesh-grants/internal/grants/part.go
+2
View File
@@ -277,6 +277,8 @@ func (p *Part) Compose() Result {
p.last = res
p.mu.Unlock()
}()
// A first view's condition the condition keeper did not take is said again at every pass until it lands.
p.Views.SayUnsaid()
in := ReadInputs(p.Read, &p.cache)
p.keepBrokers(in)
if bound, err := p.boundStreams(); err != nil {