A3g item 1: a process holder's gate-unanswered rests on a bound key, and its act names no node
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
This commit is contained in:
+5
-2
@@ -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.<name>` 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 {
|
||||
|
||||
@@ -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", "")
|
||||
|
||||
+2
-1
@@ -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)
|
||||
|
||||
+67
-22
@@ -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 {
|
||||
|
||||
+117
-11
@@ -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) {
|
||||
|
||||
@@ -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} {
|
||||
|
||||
+41
-1
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user