diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 754d1ba5..27dd3585 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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) } diff --git a/modules/mesh-grants/internal/grants/bindings.go b/modules/mesh-grants/internal/grants/bindings.go index a06c7922..47cefce0 100644 --- a/modules/mesh-grants/internal/grants/bindings.go +++ b/modules/mesh-grants/internal/grants/bindings.go @@ -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.` 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" diff --git a/modules/mesh-grants/internal/grants/firstviews.go b/modules/mesh-grants/internal/grants/firstviews.go index a9b31d75..c1dca071 100644 --- a/modules/mesh-grants/internal/grants/firstviews.go +++ b/modules/mesh-grants/internal/grants/firstviews.go @@ -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 { diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index ba5ab769..98941663 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -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 {