From f8062e268039cfe5ece97125933af222bea69614 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:42:24 +0100 Subject: [PATCH] A3g item 1: a process holder's gate-unanswered rests on a bound key, and its act names no node MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A waiting call's ask's 60 minutes are a key with its own time to live in the holder's state approvals, its end heard as the bus's marker, so no clock waits in the process; a call whose key ended while the holder was away is judged at its start. A process holder's act names no node, so two instances raise one ask. Builds: design 47 G §677, lead answer A-255 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/approval/act.go | 7 +- go/approval/closed.go | 4 +- go/approval/gate.go | 3 +- go/approval/held.go | 89 ++++++++++++++----- go/approval/held_test.go | 128 ++++++++++++++++++++++++--- go/approval/record.go | 2 +- go/process/gate.go | 42 ++++++++- go/process/gated_integration_test.go | 52 ++++++++++- 8 files changed, 286 insertions(+), 41 deletions(-) diff --git a/go/approval/act.go b/go/approval/act.go index 2d61d27e..d2e1275e 100644 --- a/go/approval/act.go +++ b/go/approval/act.go @@ -59,10 +59,13 @@ func refusedUnasked(raw json.RawMessage) string { return "" } -// ActOf is the act a marked call binds (§8): `address`, `node`, `module` and `tool` or `seat` and `verb`, `build`, and +// ActOf is the act a marked call binds (§8): `address`, `node` (none given for a process holder, lead answer A-255), `module` and `tool` or `seat` and `verb`, `build`, and // `arg.` for each argument given, a string as itself and any other value as its JSON in RFC 8785's canonical form. func ActOf(c Call, node string) (operatorask.Act, error) { - act := operatorask.Act{"address": c.Address, "node": node, "build": c.Fingerprint} + act := operatorask.Act{"address": c.Address, "build": c.Fingerprint} + if node != "" { // a process holder's act names none (lead answer A-255) + act["node"] = node + } if c.Seat == "" { act["module"], act["tool"] = c.Module, c.Name } else { diff --git a/go/approval/closed.go b/go/approval/closed.go index 00310a6d..bfd7460f 100644 --- a/go/approval/closed.go +++ b/go/approval/closed.go @@ -46,7 +46,7 @@ func (g *Gate) stateFailed(err error) { g.mu.Unlock() if first { g.d.Logf("[approval] 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 g.d.Bounds != nil && !g.d.Holding { 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("[approval] the bound of the state %s's failure is not kept: %v", StateName, e) } @@ -63,7 +63,7 @@ func (g *Gate) stateRead() { if !was { return } - if g.d.Bounds != nil { + if g.d.Bounds != nil && !g.d.Holding { _ = g.d.Bounds.Delete(g.unreadableKey()) } g.finding(StateName, KindApprovalsUnreadable, false, "the state "+StateName+" is read again", "") diff --git a/go/approval/gate.go b/go/approval/gate.go index 2f6d9cf2..16ac2efc 100644 --- a/go/approval/gate.go +++ b/go/approval/gate.go @@ -47,7 +47,8 @@ type Deps struct { Channel Channel // Warranted asks `mesh-hand-acts.warranted` {asker, ask}, each try a kept call open while open says so (§16, A-157). 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 is the runner's key-ttl state bound (A-157, A-158), or a process holder's state approvals itself (A-255); 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) diff --git a/go/approval/held.go b/go/approval/held.go index 3c0b0ff1..aef5927a 100644 --- a/go/approval/held.go +++ b/go/approval/held.go @@ -127,7 +127,7 @@ func (g *Gate) Hold(c Call, wait bool, run func() (any, error)) (any, *Held, err if why := g.held(); why != "" { return nil, nil, g.refused(c, "", why) } - act, err := ActOf(c, g.d.Node) + act, err := ActOf(c, "") // a process holder's act names no node, so two instances raise one ask (A-255) if err != nil { return nil, nil, g.refused(c, "", err.Error()) } @@ -423,7 +423,7 @@ func (g *Gate) anew(w WaitingCall) { return } w.Ask, w.Asked = id, g.d.Now().UTC() - g.armUnanswered(w) + g.rearmUnanswered(w) g.d.Go(func() { g.Decided(id) }) } @@ -454,6 +454,13 @@ func (g *Gate) say(id string) { if json.Unmarshal(e.Value, &w) != nil || w.End == nil || w.Said { return } + // Its bound key still kept says gate-unanswered was not raised, and the key goes; one the bus ended is cleared once the + // end is said (A-255). + b, err := g.d.State.Get(unansweredHeldKey(id)) + raised := err == nil && b == nil + if b != nil && g.d.Bounds != nil { + _ = g.d.Bounds.Delete(unansweredHeldKey(id)) + } if g.d.Say == nil { g.d.Logf("[approval] the end %s of call %s is not said: the holder says no gated-ended", w.End.Outcome, id) return @@ -464,7 +471,7 @@ func (g *Gate) say(id string) { } g.d.Logf("[approval] %s by %s ended %s (call %s, ask %s)", w.Address, w.Caller, w.End.Outcome, id, w.Ask) _ = g.updateCall(id, func(k *WaitingCall) { k.Said = true }) - if g.d.Now().Sub(w.Asked) >= AskLasts && g.d.Findings != nil { + if raised && g.d.Findings != nil { _ = g.d.Findings.Clear(nilCtx(), KindGateUnanswered, id, "the call's end, "+w.End.Outcome+", is said") } } @@ -529,26 +536,61 @@ func (g *Gate) calls() ([]WaitingCall, error) { return out, nil } -// armUnanswered raises the waiting call's gate-unanswered when its ask's 60 minutes pass with no end of it (§677). +// unansweredHeldKey is a waiting call's bound key in the holder's state approvals (lead answer A-255): created with the +// ask's 60 minutes as its own time to live, its expiry heard as the bus's event raises gate-unanswered. +func unansweredHeldKey(call string) string { return "unanswered." + call } + +// armUnanswered keeps the waiting call's bound key (§677; A-255): when its ask's 60 minutes pass with no end of it, the bus +// ends the key and the holder, hearing that, raises gate-unanswered; no clock waits in the process. A call already past its +// 60 minutes (its key ended while the holder was away) is judged at once. func (g *Gate) armUnanswered(w WaitingCall) { - left := max(w.Asked.Add(AskLasts).Sub(g.d.Now()), 0) - at := g.d.After(left) - go func() { // its own goroutine whatever Go is: it waits out the ask's 60 minutes on the clock's pause - <-at - e, err := g.d.State.Get(callKey(w.ID)) - if err != nil || e == nil { - return - } - var now WaitingCall - if json.Unmarshal(e.Value, &now) != nil || now.End != nil || now.Ask != w.Ask || g.d.Findings == nil { - return - } - if err := g.d.Findings.Raise(nilCtx(), KindGateUnanswered, w.ID, fmt.Sprintf("the call %s of %s by %s waits on ask %s, "+ - "past its %s with no decided heard: nothing ran; the holder's %s.approvals reads it", w.ID, w.Address, w.Caller, w.Ask, - AskLasts, w.Seat), "a call waits on an unanswered ask"); err != nil { - g.d.Logf("[approval] the finding %s of call %s is not kept: %v", KindGateUnanswered, w.ID, err) - } - }() + left := w.Asked.Add(AskLasts).Sub(g.d.Now()) + if left <= 0 { + g.heldUnanswered(w.ID) + return + } + if g.d.Bounds == nil { + g.d.Logf("[approval] the call %s waits on ask %s with no bound kept: the holder's %s keeps no key with its own time to live", + w.ID, w.Ask, StateName) + return + } + if err := g.d.Bounds.CreateTTL(unansweredHeldKey(w.ID), map[string]string{"call": w.ID, "ask": w.Ask}, left); err != nil && !isChanged(err) { + g.d.Logf("[approval] the bound of call %s on ask %s is not kept: %v", w.ID, w.Ask, err) + } +} + +// rearmUnanswered keeps a call asked anew under a new bound of its new ask's 60 minutes. +func (g *Gate) rearmUnanswered(w WaitingCall) { + if g.d.Bounds != nil { + _ = g.d.Bounds.Delete(unansweredHeldKey(w.ID)) + } + g.armUnanswered(w) +} + +// heldUnanswered is a waiting call's bound ended (A-255): gate-unanswered raised while the call has no end. +func (g *Gate) heldUnanswered(id string) { + e, err := g.d.State.Get(callKey(id)) + if err != nil || e == nil { + return + } + var w WaitingCall + if json.Unmarshal(e.Value, &w) != nil || w.End != nil || g.d.Findings == nil { + return + } + if err := g.d.Findings.Raise(nilCtx(), KindGateUnanswered, w.ID, fmt.Sprintf("the call %s of %s by %s waits on ask %s, "+ + "past its %s with no decided heard: nothing ran; the holder's %s.approvals reads it", w.ID, w.Address, w.Caller, w.Ask, + AskLasts, w.Seat), "a call waits on an unanswered ask"); err != nil { + g.d.Logf("[approval] the finding %s of call %s is not kept: %v", KindGateUnanswered, w.ID, err) + } +} + +// watchHeld hears the waiting calls' bound keys end (A-255) for the life of the holder's connection. +func (g *Gate) watchHeld() error { + if g.d.Bounds == nil { + return nil + } + _, err := g.d.Bounds.Watch("unanswered.*", func(key string) { g.heldUnanswered(strings.TrimPrefix(key, "unanswered.")) }) + return err } // StartHolding is the holder's read at its start (§677): every waiting call is read, each ask already decided acted on as @@ -561,6 +603,9 @@ func (g *Gate) StartHolding() error { return fmt.Errorf("the waiting calls in %s cannot be read: %w", StateName, err) } g.stateRead() + if err := g.watchHeld(); err != nil { + return fmt.Errorf("the bound keys in %s cannot be watched: %w", StateName, err) + } asks := map[string]bool{} for _, w := range all { switch { diff --git a/go/approval/held_test.go b/go/approval/held_test.go index 81cd9c81..e57779c1 100644 --- a/go/approval/held_test.go +++ b/go/approval/held_test.go @@ -80,6 +80,56 @@ func (m *memStore) Keys() ([]string, error) { return out, nil } +// bounds is the holder's bound keys in its state approvals as a fake (A-255): each key kept in the state with its time to +// live, expire ending it as the bus does and handing it to the watch. +type bounds struct { + state *memStore + mu sync.Mutex + ttl map[string]time.Duration + watched func(string) +} + +func (b *bounds) CreateTTL(key string, value any, ttl time.Duration) error { + if _, err := b.state.Create(key, value); err != nil { + return err + } + b.mu.Lock() + defer b.mu.Unlock() + b.ttl[key] = ttl + return nil +} +func (b *bounds) Delete(key string) error { + b.state.delete(key) + b.mu.Lock() + defer b.mu.Unlock() + delete(b.ttl, key) + return nil +} +func (b *bounds) Watch(pattern string, expired func(string)) (func(), error) { + b.mu.Lock() + defer b.mu.Unlock() + b.watched = expired + return func() {}, nil +} +func (b *bounds) expire(key string) { + b.mu.Lock() + _, ok := b.ttl[key] + delete(b.ttl, key) + w := b.watched + b.mu.Unlock() + b.state.delete(key) + if ok && w != nil { + w(key) + } +} + +func (m *memStore) delete(key string) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.vals, key) + delete(m.revs, key) +} + // router is the operator channel as a fake: it keeps each ask the gate raises, and the test decides its record. type router struct { mu sync.Mutex @@ -172,9 +222,10 @@ func TestAHolderHoldsAMarkedCall(t *testing.T) { recs *int finds *findings after chan time.Time + bounds *bounds } - newGate := func(state *memStore, r *router, ran, recs *int, said map[string]GatedEnded, after chan time.Time, f *findings) *Gate { - g := New(Deps{Node: "n1", Runner: "holder", Holding: true, Build: "fp1", State: state, Channel: r, Findings: f, + newGate := func(state *memStore, r *router, ran, recs *int, said map[string]GatedEnded, after chan time.Time, f *findings, b *bounds) *Gate { + g := New(Deps{Node: "n1", Runner: "holder", Holding: true, Build: "fp1", State: state, Channel: r, Findings: f, Bounds: b, Now: func() time.Time { return now }, Go: func(f func()) { f() }, After: func(d time.Duration) <-chan time.Time { if d < time.Minute { // the record's short pauses pass at once @@ -200,8 +251,8 @@ func TestAHolderHoldsAMarkedCall(t *testing.T) { r := &router{state: state, records: map[string]*Record{}} ran, recs := 0, 0 w := world{state: state, router: r, said: map[string]GatedEnded{}, ran: &ran, recs: &recs, - finds: &findings{raised: map[string]bool{}}, after: make(chan time.Time)} - w.g = newGate(state, r, w.ran, w.recs, w.said, w.after, w.finds) + finds: &findings{raised: map[string]bool{}}, after: make(chan time.Time), bounds: &bounds{state: state, ttl: map[string]time.Duration{}}} + w.g = newGate(state, r, w.ran, w.recs, w.said, w.after, w.finds, w.bounds) return w } call := func(caller, args string) Call { @@ -285,7 +336,7 @@ func TestAHolderHoldsAMarkedCall(t *testing.T) { if _, h, err := hold(t, w, call("n1.a", `{"token":"ghp_0123456789abcdefghijklmnopqrstuvwxyzAB"}`), true); err == nil || h != nil { t.Fatalf("held %+v, %v", h, err) } - if keys, _ := w.state.Keys(); len(keys) != 0 || len(w.router.asks) != 0 { + if keys, _ := w.state.Keys(); len(keys) != 0 || len(w.router.asks) != 0 || len(w.bounds.ttl) != 0 { t.Errorf("kept %v, asked %d", keys, len(w.router.asks)) } }) @@ -305,7 +356,7 @@ func TestAHolderHoldsAMarkedCall(t *testing.T) { w := setup() _, h, _ := hold(t, w, call("n1.a", `{}`), true) w.router.choose(h.Asked, OptionApprove, now) - again := newGate(w.state, w.router, w.ran, w.recs, w.said, w.after, w.finds) + again := newGate(w.state, w.router, w.ran, w.recs, w.said, w.after, w.finds, w.bounds) if err := again.StartHolding(); err != nil || *w.ran != 1 || ended(w, "n1.a", h.Call).Outcome != OutcomeRan { t.Fatalf("start: %v, ran %d", err, *w.ran) } @@ -314,13 +365,68 @@ func TestAHolderHoldsAMarkedCall(t *testing.T) { t.Errorf("ran %d after a second start", *w.ran) } }) - t.Run("past its ask's 60 minutes with no decided, gate-unanswered", func(t *testing.T) { + t.Run("past its ask's 60 minutes with no decided, gate-unanswered on its bound key's end", func(t *testing.T) { + // A-255: a bound key with the ask's 60 minutes to live in the state approvals, never a clock wait. w := setup() - w.finds.heard = make(chan string, 1) + if err := w.g.StartHolding(); err != nil { + t.Fatal(err) + } _, h, _ := hold(t, w, call("n1.a", `{}`), true) - w.after <- now - if got := <-w.finds.heard; got != KindGateUnanswered+" "+h.Call { - t.Errorf("raised %s", got) + key := "unanswered." + h.Call + if ttl := w.bounds.ttl[key]; ttl != AskLasts { + t.Fatalf("bound %s kept for %s (%v)", key, ttl, w.bounds.ttl) + } + if w.finds.raised[KindGateUnanswered+" "+h.Call] { + t.Fatal("raised before its bound ended") + } + w.bounds.expire(key) + if !w.finds.raised[KindGateUnanswered+" "+h.Call] { + t.Errorf("not raised on its bound's end: %v", w.finds.raised) + } + w.router.choose(h.Asked, OptionDecline, now) + w.g.Decided(h.Asked) + if w.finds.raised[KindGateUnanswered+" "+h.Call] { + t.Error("not cleared once its end is said") + } + }) + t.Run("a call whose bound ended while the holder was away is judged at its start", func(t *testing.T) { + w := setup() + _, h, _ := hold(t, w, call("n1.a", `{}`), true) + w.bounds.watched = nil + w.bounds.expire("unanswered." + h.Call) // the bus ended it with nobody watching + now = now.Add(AskLasts + time.Minute) + defer func() { now = now.Add(-AskLasts - time.Minute) }() + again := newGate(w.state, w.router, w.ran, w.recs, w.said, w.after, w.finds, w.bounds) + if err := again.StartHolding(); err != nil || !w.finds.raised[KindGateUnanswered+" "+h.Call] { + t.Errorf("start: %v, raised %v", err, w.finds.raised) + } + }) + t.Run("two instances of the holder raise one ask and run once", func(t *testing.T) { + // A-255: a process holder's act names no node. + w := setup() + other := New(Deps{Node: "n2", Runner: "holder", Holding: true, Build: "fp1", State: w.state, Channel: w.router, + Findings: w.finds, Bounds: w.bounds, Now: func() time.Time { return now }, Go: func(f func()) { f() }, + Say: func(subject string, body any, id string) error { w.said[subject] = body.(GatedEnded); return nil }}) + other.Serves("wiper", "wipe", func(string, json.RawMessage) (any, error) { *w.ran++; return "wiped", nil }) + _, h1, _ := hold(t, w, call("n1.a", `{}`), true) + _, h2, err := other.Hold(call("n2.b", `{}`), true, nil) + if err != nil || h2 == nil || h2.Asked != h1.Asked || len(w.router.asks) != 1 { + t.Fatalf("second instance: %+v, %v; asks %d", h2, err, len(w.router.asks)) + } + if strings.Contains(w.router.asks[0].About, "n1") { + t.Errorf("the ask binds %s", w.router.asks[0].About) + } + var first WaitingCall + e, _ := w.state.Get("call." + h1.Call) + _ = json.Unmarshal(e.Value, &first) + if _, ok := first.Act["node"]; ok { + t.Errorf("the act names a node: %v", first.Act) + } + w.router.choose(h1.Asked, OptionApprove, now) + w.g.Decided(h1.Asked) + other.Decided(h1.Asked) + if *w.ran != 1 { + t.Errorf("ran %d", *w.ran) } }) t.Run("a state that cannot be read refuses every marked call", func(t *testing.T) { diff --git a/go/approval/record.go b/go/approval/record.go index 97dc8a44..aaf74be0 100644 --- a/go/approval/record.go +++ b/go/approval/record.go @@ -62,7 +62,7 @@ func (g *Gate) ran(c Call, id string, rec *Record, answer any, err error) (any, // bindRecording keeps the run's two bound keys: one that raises `call-unanswered` when `warranted` has not answered within // five minutes, one that says an unrecorded run an hour before its key's 48 hours end (§13, §16). func (g *Gate) bindRecording(module, id string) { - if g.d.Bounds == nil { + if g.d.Bounds == nil || g.d.Holding { // a process holder's bound keys are its waiting calls' alone (A-255) return } for key, ttl := range map[string]time.Duration{g.unansweredKey(id): UnansweredAfter, g.unrecordedKey(id): KeptFor - SaidBefore} { diff --git a/go/process/gate.go b/go/process/gate.go index f4c372fb..3d6e5733 100644 --- a/go/process/gate.go +++ b/go/process/gate.go @@ -36,8 +36,9 @@ func (c *Conn) Gate(g Gating) *approval.Gate { if build == "" { build = OwnBuild().Fingerprint } + state := c.State(approval.StateName) return approval.New(approval.Deps{Node: c.account.Node, Runner: c.account.Module, Holding: true, Build: build, - State: c.State(approval.StateName), Channel: g.Channel, Findings: g.Findings, Logf: g.Logf, + State: state, Bounds: heldBounds{state}, Channel: g.Channel, Findings: g.Findings, Logf: g.Logf, Warranted: func(asker, ask string, open func() (bool, error)) error { _, err := c.CallKeptWhileOpen(busretry.Open{Key: ask, Open: func(string) (bool, error) { return open() }, Logf: g.Logf}, "seat:mesh-hand-acts.warranted", map[string]string{"asker": asker, "ask": ask}) @@ -48,6 +49,45 @@ func (c *Conn) Gate(g Gating) *approval.Gate { }}) } +// heldBounds is the holder's bound keys in its own state approvals (lead answer A-255): a key with its own time to live, +// whose end the bus says as a marker that a watch hears, so a waiting call's 60 minutes need no clock in the process. +type heldBounds struct{ s State } + +func (b heldBounds) CreateTTL(key string, value any, ttl time.Duration) error { + _, err := b.s.CreateTTL(key, value, ttl) + return err +} + +func (b heldBounds) Delete(key string) error { return b.s.Delete(key) } + +// Watch hands expired each key under pattern whose own time to live ended, heard from now on: the bus's marker of a time +// to live is a purge, which this state's own writes never make (a delete is not one). +func (b heldBounds) Watch(pattern string, expired func(key string)) (func(), error) { + ctx, cancel := context.WithCancel(context.Background()) + bind, stop := context.WithTimeout(ctx, stateTimeout) + kv, err := b.s.store(bind) + stop() + if err != nil { + cancel() + return nil, err + } + w, err := kv.Watch(ctx, watchPattern(pattern), jetstream.UpdatesOnly()) + if err != nil { + cancel() + return nil, fmt.Errorf("%s's state %s, watching %q: %w", b.s.Module, b.s.Name, pattern, err) + } + go func() { + defer cancel() + defer func() { _ = w.Stop() }() + for e := range w.Updates() { + if e != nil && e.Operation() == jetstream.KeyValuePurge { + expired(e.Key()) + } + } + }() + return cancel, nil +} + // HearDecided hands the gate each `decided` of the holder's own asks that hub hears (§677): only the ask's id is read // from it, the gate reading the router's own record. func HearDecided(g *approval.Gate, ch approval.Channel, module string, hub *EventHub) error { diff --git a/go/process/gated_integration_test.go b/go/process/gated_integration_test.go index 5ddad2dc..9a41f58b 100644 --- a/go/process/gated_integration_test.go +++ b/go/process/gated_integration_test.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/operatorask" "git.novox.be/novox/mesh-sdk/go/seatdef" @@ -98,7 +99,8 @@ func TestAProcessHoldersMarkedVerbIsAskedApprovedRunAndRecorded(t *testing.T) { if err != nil { t.Fatal(err) } - if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "holder_" + approval.StateName, TTL: 48 * time.Hour}); err != nil { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "holder_" + approval.StateName, TTL: 48 * time.Hour, + LimitMarkerTTL: time.Minute}); err != nil { t.Fatal(err) } if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.operator-channel.event.>"}}); err != nil { @@ -238,6 +240,54 @@ func TestAProcessHoldersMarkedVerbIsAskedApprovedRunAndRecorded(t *testing.T) { t.Errorf("ran %d", ran) } }) + t.Run("a waiting call's bound key is kept with its time to live, and its end is heard", func(t *testing.T) { + // A-255: gate-unanswered rests on a bound key in the state approvals, never a clock wait. + state := holder.State(approval.StateName) + kept := make(chan string, 1) + if err := state.Watch("unanswered.*", func(c stdio.StateChange) error { + if c.Op == stdio.OpPut && !c.Current { + kept <- c.Key + } + return nil + }); err != nil { + t.Fatal(err) + } + done := callWipe(map[string]string{"dir": "bound"}) + a := asked() + var bound string + select { + case bound = <-kept: + case <-time.After(10 * time.Second): + t.Fatal("no bound key kept within 10s") + } + decide(a, approval.OptionDecline) + wait(done) + if e, _ := state.Get(bound); e != nil { + t.Errorf("the bound %s of an ended call is kept", bound) + } + heard := make(chan string, 2) + stop, err := heldBounds{state}.Watch("unanswered.*", func(k string) { heard <- k }) + if err != nil { + t.Fatal(err) + } + defer stop() + b := heldBounds{state} + if err := b.CreateTTL("unanswered.gone", map[string]string{}, MinTTL); err != nil { + t.Fatal(err) + } + if err := b.CreateTTL("unanswered.deleted", map[string]string{}, time.Hour); err != nil { + t.Fatal(err) + } + _ = b.Delete("unanswered.deleted") + select { + case k := <-heard: + if k != "unanswered.gone" { + t.Errorf("heard %s", k) + } + case <-time.After(10 * time.Second): + t.Fatal("the bound key's end was not heard within 10s") + } + }) t.Run("failing closed: no gate, or no state approvals", func(t *testing.T) { if _, err := holder.AnswerSeat("bare", marked("bare"), wipe); err != nil { t.Fatal(err)