Merge pull request 'An asker of the build seats hears when its ask is given up, and the seat's users may read their own dead letters (issue 734)' (#92) from fix/734-lost-ask into main
This commit was merged in pull request #92.
This commit is contained in:
@@ -69,6 +69,37 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
// An asker on a seat pulled by kind hears what that seat's workers gave up on, and no other stream's dead letters
|
||||
// (novox/hq issue 734: a check ask given up during a bus reload left its try running for ever).
|
||||
t.Run("an asker reads its seat's dead letters", func(t *testing.T) {
|
||||
build := Seat{Name: "node-build-go", Scope: "node", Accepts: []string{"delivery", "check", "recheck"}, PulledByKind: true}
|
||||
perms, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-delivery-checker", Uses: []Seat{build}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for subject, want := range map[string]bool{
|
||||
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.>": true,
|
||||
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.SEAT_NODE_BUILD_GO.w.next": true,
|
||||
"$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1": true,
|
||||
"$JS.API.CONSUMER.MSG.NEXT.DEAD_LETTERS.x1": true,
|
||||
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1.mesh.events.dead.EVENTS.>": false,
|
||||
"$JS.API.CONSUMER.CREATE.DEAD_LETTERS.x1": false,
|
||||
"$JS.API.STREAM.PURGE.DEAD_LETTERS": false,
|
||||
"$JS.FC.DEAD_LETTERS.x1": false,
|
||||
} {
|
||||
if got := MayPublish(perms, subject); got != want {
|
||||
t.Errorf("may publish %s: %v, want %v", subject, got, want)
|
||||
}
|
||||
}
|
||||
plain, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "shop",
|
||||
Uses: []Seat{{Name: "telegram-sender", Accepts: []string{"send"}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if MayPublish(plain, "$JS.API.CONSUMER.DELETE.DEAD_LETTERS.x1") {
|
||||
t.Error("a user of a seat not pulled by kind reads DEAD_LETTERS")
|
||||
}
|
||||
})
|
||||
t.Run("the controller's golden server configuration", func(t *testing.T) {
|
||||
seat := Seat{Name: "telegram-sender", Accepts: []string{"send"}, Emits: []string{"delivered", "failed"}}
|
||||
got, err := Compose(
|
||||
|
||||
@@ -24,8 +24,9 @@ import (
|
||||
// subjects cover it, so what travels there — a code typed by the operator — is never persisted. A kinded
|
||||
// holder asks with its own kind; the modules that watch the seat answer.
|
||||
//
|
||||
// And one read: **a holder's records**, a bucket the seat names, read by each user under its own name only
|
||||
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's.
|
||||
// And two reads: **a holder's records**, a bucket the seat names, read by each user under its own name only
|
||||
// (`$KV.<bucket>.<module>.>`), so an asker reads the state of its own asks and no other asker's; and on a seat pulled by
|
||||
// kind, **what its workers gave up on**, the copies DEAD_LETTERS keeps of the seat's stream (DeadLetterReads, issue 734).
|
||||
|
||||
// BuildAskMaxDeliver bounds the deliveries of a build seat's ask (issue 1907): mesh-sdk delivery.AskMaxDeliver, written
|
||||
// here as the broker mirrors the catalogue rather than importing it.
|
||||
@@ -129,6 +130,9 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
|
||||
}
|
||||
t.Records = append(t.Records, "$JS.API.DIRECT.GET.KV_"+b+".$KV."+b+"."+module+".>")
|
||||
}
|
||||
if s.PulledByKind && len(s.Accepts) > 0 && safeSubject.MatchString(s.Name) {
|
||||
t.Records = append(t.Records, DeadLetterReads(seatStreamName(s.Name))...)
|
||||
}
|
||||
}
|
||||
for _, w := range watches {
|
||||
if w.Kinded {
|
||||
@@ -151,6 +155,20 @@ func SeatTrafficOf(module string, holds, uses, watches []Seat) SeatTraffic {
|
||||
return t
|
||||
}
|
||||
|
||||
// DeadLetterReads is what a user of a seat pulled by kind is granted to hear which of the asks it put the seat's workers
|
||||
// gave up on (novox/hq issue 734): an ordered consumer on DEAD_LETTERS filtered to the copies the controller keeps of that
|
||||
// seat's stream (DeadLetterSubject(stream, *)), pulled, and its deletion. A copy carries the ask's own message id
|
||||
// (Mesh-Dead-Msg-Id), so the asker ends the step it waits for instead of waiting for ever; it reads the seat's dead
|
||||
// letters and no other stream's, and changes none.
|
||||
func DeadLetterReads(stream string) []string {
|
||||
return []string{
|
||||
"$JS.API.STREAM.INFO." + DeadLettersStream,
|
||||
"$JS.API.CONSUMER.CREATE." + DeadLettersStream + ".*." + DeadLetterSubject(stream, ">"),
|
||||
"$JS.API.CONSUMER.MSG.NEXT." + DeadLettersStream + ".*",
|
||||
"$JS.API.CONSUMER.DELETE." + DeadLettersStream + ".*",
|
||||
}
|
||||
}
|
||||
|
||||
// seatTrafficGrants is the bus permissions seat traffic needs: the subjects themselves, and the JetStream API a
|
||||
// worker is pulled and acknowledged through and a record is read through.
|
||||
func seatTrafficGrants(t SeatTraffic) (pub, sub []string) {
|
||||
|
||||
@@ -2,11 +2,15 @@ package buildrun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
func goAsk(id string) delivery.StepAsk {
|
||||
@@ -48,6 +52,74 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
|
||||
if len(g.Asks()) != 0 {
|
||||
t.Fatal("a wait put an ask")
|
||||
}
|
||||
// Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed
|
||||
// with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written
|
||||
// first is answered as the node wrote it.
|
||||
dead := &DeadLetters{}
|
||||
lw := LostAskWaiter{Inner: g, Dead: dead}
|
||||
k = Asker{Publisher: g, Waiter: lw, Bound: time.Minute}
|
||||
g.Write(delivery.Step{ID: "mvl0frv18d1970", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
|
||||
given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"},
|
||||
"Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}})
|
||||
if !ok {
|
||||
t.Fatal("a dead letter's headers were not read")
|
||||
}
|
||||
go dead.Heard(given)
|
||||
step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970"))
|
||||
if err != nil || step.State != delivery.StepFailed || !IsLost(step) ||
|
||||
!strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") {
|
||||
t.Fatalf("a given-up ask: %+v, %v", step, err)
|
||||
}
|
||||
if _, err := k.Passed(step, nil); err == nil || !strings.Contains(err.Error(), "on "+LostNode+" failed: "+LostWords) {
|
||||
t.Fatalf("a given-up ask's error: %v", err)
|
||||
}
|
||||
step, err = k.Wait(NotTakenBound(context.Background(), 20*time.Millisecond), goAsk("mvl0frv18d1971"))
|
||||
if err != nil || !IsNotTaken(step) || IsLost(step) {
|
||||
t.Fatalf("an ask no node took: %+v, %v", step, err)
|
||||
}
|
||||
g.Write(delivery.Step{ID: "mvl0frv18d1972", Node: "g14", State: delivery.StepPassed})
|
||||
if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" {
|
||||
t.Fatalf("a step written: %+v, %v", step, err)
|
||||
}
|
||||
// Review round 1 of issue 734: a hearing the bus refuses at start is deaf and says so, never holds the start up, and
|
||||
// tries again on the next ask waited for; heard, it says so and hands the given-up asks on.
|
||||
fol := &refusingFollower{refuse: 1}
|
||||
var mu sync.Mutex
|
||||
var told []Hearing
|
||||
hearing := NewDeadLetterHearing(context.Background(), fol, []string{delivery.AskGo}, dead, nil,
|
||||
func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() })
|
||||
hearing.Wait()
|
||||
if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") {
|
||||
t.Fatalf("a refused hearing says %q", why)
|
||||
}
|
||||
k = Asker{Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute}
|
||||
g.Write(delivery.Step{ID: "mvl0frv18d1973", Node: "g14", State: delivery.StepPassed})
|
||||
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hearing.Wait()
|
||||
mu.Lock()
|
||||
if hearing.Deaf() != "" || len(told) != 2 || told[0].Heard || !told[1].Heard || fol.calls() != 2 {
|
||||
t.Fatalf("after an ask: deaf %q, told %+v, followed %d times", hearing.Deaf(), told, fol.calls())
|
||||
}
|
||||
mu.Unlock()
|
||||
fol.heard(GivenUp{AskID: "mvl0frv18d1974", Deliveries: 5})
|
||||
step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1974"))
|
||||
if err != nil || !IsLost(step) {
|
||||
t.Fatalf("an ask heard given up after the hearing came back: %+v, %v", step, err)
|
||||
}
|
||||
for _, row := range []struct {
|
||||
h *Hearing
|
||||
want string
|
||||
}{{nil, health.StateUnhealthy}, {&told[0], health.StateUnhealthy}, {&told[1], health.StateHealthy}} {
|
||||
var raw []byte
|
||||
if row.h != nil {
|
||||
raw, _ = json.Marshal(row.h)
|
||||
}
|
||||
if a := HearingHealth(raw); a.State != row.want || a.Why == "" {
|
||||
t.Errorf("health of %+v = %+v, want %s", row.h, a, row.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the
|
||||
@@ -69,3 +141,26 @@ func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
|
||||
t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err)
|
||||
}
|
||||
}
|
||||
|
||||
// refusingFollower refuses the first `refuse` follows, as the bus does before the asker's grant is composed.
|
||||
type refusingFollower struct {
|
||||
mu sync.Mutex
|
||||
refuse int
|
||||
n int
|
||||
hand func(GivenUp)
|
||||
}
|
||||
|
||||
func (f *refusingFollower) Follow(_ context.Context, _ string, _ time.Time, heard func(GivenUp)) (func(), error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.n++
|
||||
if f.n <= f.refuse {
|
||||
return nil, errors.New("permissions violation")
|
||||
}
|
||||
f.hand = heard
|
||||
return func() {}, nil
|
||||
}
|
||||
|
||||
func (f *refusingFollower) calls() int { f.mu.Lock(); defer f.mu.Unlock(); return f.n }
|
||||
|
||||
func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unlock(); h(g) }
|
||||
|
||||
@@ -0,0 +1,495 @@
|
||||
package buildrun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask
|
||||
// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node
|
||||
// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under
|
||||
// mesh.events.dead.<seat stream>.<worker>, with the ask's message id (its ask id), how often it was handed out and when it
|
||||
// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays
|
||||
// running and holds a try place, a check's verdict stays running and its pull request waits.
|
||||
//
|
||||
// An asker of the build seats (the checker, the builder) hears those copies (DeadLetterHearing: an ordered consumer on
|
||||
// DEAD_LETTERS filtered to the build seats' streams, granted to a user of a seat pulled by kind, buscompose
|
||||
// DeadLetterReads) into DeadLetters, and LostAskWaiter ends the waiting step at once as failed on the node LostNode, its
|
||||
// reason saying LostWords; the asker decides what follows (the checker ends a try as error and asks a check again, the
|
||||
// builder asks its part of a build again).
|
||||
//
|
||||
// Belt and braces, where the asker says so in the context (NotTakenBound; the checker's tries): an ask no build node took
|
||||
// within a bound ends the same way, saying NotTakenWords. An ask waits in the stream as long as the backlog is otherwise
|
||||
// (ADR 0525 §9): a backlog is no fault.
|
||||
|
||||
// LostNode is the node a step ended for a lost ask names: no build node wrote it.
|
||||
const LostNode = "bus"
|
||||
|
||||
// The words of a step ended for its ask.
|
||||
const (
|
||||
LostWords = "its ask was given up by the build seat"
|
||||
NotTakenWords = "no build node took its ask"
|
||||
)
|
||||
|
||||
// The headers the controller keeps a given-up message's copy with in DEAD_LETTERS (mesh-controller internal/link
|
||||
// deadletters.go): the original message id (an ask's is its ask id), how often it was handed out, when it was given up.
|
||||
const (
|
||||
DeadHeaderMsgID = "Mesh-Dead-Msg-Id"
|
||||
DeadHeaderDeliveries = "Mesh-Dead-Deliveries"
|
||||
DeadHeaderGaveUp = "Mesh-Dead-Gave-Up"
|
||||
DeadHeaderStream = "Mesh-Dead-Stream"
|
||||
)
|
||||
|
||||
// DeadLettersStream is the stream the controller keeps given-up messages in, and DeadLetterFilter the subjects of one seat
|
||||
// stream's copies there (buscompose DeadLetterSubject).
|
||||
const DeadLettersStream = "DEAD_LETTERS"
|
||||
|
||||
// DeadLetterFilter is the subjects DEAD_LETTERS keeps a seat stream's given-up messages under.
|
||||
func DeadLetterFilter(stream string) string { return "mesh.events.dead." + stream + ".>" }
|
||||
|
||||
// SeatStream is the stream of a build seat's language (buscompose seatStreamName of node-build-<language>).
|
||||
func SeatStream(language string) string { return "SEAT_NODE_BUILD_" + strings.ToUpper(language) }
|
||||
|
||||
// GivenUp is one ask a build seat gave up on, as DEAD_LETTERS keeps it.
|
||||
type GivenUp struct {
|
||||
AskID string
|
||||
Stream string
|
||||
Deliveries int
|
||||
At time.Time
|
||||
}
|
||||
|
||||
// GivenUpOf reads a DEAD_LETTERS copy's headers; false when it names no message id.
|
||||
func GivenUpOf(header map[string][]string) (GivenUp, bool) {
|
||||
get := func(k string) string {
|
||||
if v := header[k]; len(v) > 0 {
|
||||
return strings.TrimSpace(v[0])
|
||||
}
|
||||
return ""
|
||||
}
|
||||
g := GivenUp{AskID: get(DeadHeaderMsgID), Stream: get(DeadHeaderStream)}
|
||||
if g.AskID == "" {
|
||||
return GivenUp{}, false
|
||||
}
|
||||
g.Deliveries, _ = strconv.Atoi(get(DeadHeaderDeliveries))
|
||||
g.At, _ = time.Parse(time.RFC3339Nano, get(DeadHeaderGaveUp))
|
||||
return g, true
|
||||
}
|
||||
|
||||
// Words are what a step ended for the given-up ask says.
|
||||
func (g GivenUp) Words() string {
|
||||
at := "at an unknown time"
|
||||
if !g.At.IsZero() {
|
||||
at = "at " + g.At.UTC().Format(time.RFC3339)
|
||||
}
|
||||
return fmt.Sprintf("%s after %d deliveries %s (kept in %s); ask again", LostWords, g.Deliveries, at, DeadLettersStream)
|
||||
}
|
||||
|
||||
// keptGivenUp is how many given-up asks DeadLetters remembers for a waiter that starts after the copy was heard.
|
||||
const keptGivenUp = 1024
|
||||
|
||||
// DeadLetters holds the given-up asks heard, and hands each to the waiter of its ask. Safe for concurrent use.
|
||||
type DeadLetters struct {
|
||||
mu sync.Mutex
|
||||
heard map[string]GivenUp
|
||||
order []string
|
||||
watchers map[string][]chan GivenUp
|
||||
}
|
||||
|
||||
// Heard takes one given-up ask in.
|
||||
func (d *DeadLetters) Heard(g GivenUp) {
|
||||
d.mu.Lock()
|
||||
if d.heard == nil {
|
||||
d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{}
|
||||
}
|
||||
if _, ok := d.heard[g.AskID]; !ok {
|
||||
d.order = append(d.order, g.AskID)
|
||||
if len(d.order) > keptGivenUp {
|
||||
delete(d.heard, d.order[0])
|
||||
d.order = d.order[1:]
|
||||
}
|
||||
}
|
||||
d.heard[g.AskID] = g
|
||||
ws := d.watchers[g.AskID]
|
||||
delete(d.watchers, g.AskID)
|
||||
d.mu.Unlock()
|
||||
for _, w := range ws {
|
||||
w <- g // buffered, one each
|
||||
}
|
||||
}
|
||||
|
||||
// watch answers a channel that gets the ask's given-up copy once (at once when it was heard already), and its cancel.
|
||||
func (d *DeadLetters) watch(askID string) (<-chan GivenUp, func()) {
|
||||
ch := make(chan GivenUp, 1)
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if d.heard == nil {
|
||||
d.heard, d.watchers = map[string]GivenUp{}, map[string][]chan GivenUp{}
|
||||
}
|
||||
if g, ok := d.heard[askID]; ok {
|
||||
ch <- g
|
||||
return ch, func() {}
|
||||
}
|
||||
d.watchers[askID] = append(d.watchers[askID], ch)
|
||||
return ch, func() {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
ws := d.watchers[askID]
|
||||
for i, w := range ws {
|
||||
if w == ch {
|
||||
d.watchers[askID] = append(ws[:i], ws[i+1:]...)
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(d.watchers[askID]) == 0 {
|
||||
delete(d.watchers, askID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type notTakenKey struct{}
|
||||
|
||||
// NotTakenBound says, in ctx, that an ask no build node took within bound ends its step (a try's, never a check's).
|
||||
func NotTakenBound(ctx context.Context, bound time.Duration) context.Context {
|
||||
return context.WithValue(ctx, notTakenKey{}, bound)
|
||||
}
|
||||
|
||||
func notTakenBoundOf(ctx context.Context) time.Duration {
|
||||
d, _ := ctx.Value(notTakenKey{}).(time.Duration)
|
||||
return d
|
||||
}
|
||||
|
||||
// LostAskWaiter is an AskWaiter that ends an ask's step when the build seat gave the ask up (DeadLetters), or, under
|
||||
// NotTakenBound, when no build node took it within the bound: it hands the asker a failed step whose reason says why, on
|
||||
// the node "bus". Every step the build module writes is handed on as it is.
|
||||
type LostAskWaiter struct {
|
||||
Inner AskWaiter
|
||||
Dead *DeadLetters
|
||||
// Hearing, when set, is asked to hear DEAD_LETTERS again each time an ask is waited for while it cannot: an ask put is
|
||||
// the event that makes hearing matter.
|
||||
Hearing interface {
|
||||
Deaf() string
|
||||
Again()
|
||||
}
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func (w LostAskWaiter) now() time.Time {
|
||||
if w.Now != nil {
|
||||
return w.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
// WatchAsk is AskWaiter's.
|
||||
func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
|
||||
if w.Hearing != nil && w.Hearing.Deaf() != "" {
|
||||
w.Hearing.Again()
|
||||
}
|
||||
inner, err := w.Inner.WatchAsk(ctx, module, askID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var dead <-chan GivenUp
|
||||
cancel := func() {}
|
||||
if w.Dead != nil {
|
||||
dead, cancel = w.Dead.watch(askID)
|
||||
}
|
||||
// The not-taken bound is a context's (ADR 0437 §8: no timer), ended when a node takes the ask.
|
||||
var notTaken <-chan struct{}
|
||||
bound := notTakenBoundOf(ctx)
|
||||
stopBound := func() {}
|
||||
if bound > 0 {
|
||||
var nt context.Context
|
||||
nt, stopBound = context.WithTimeout(context.Background(), bound)
|
||||
notTaken = nt.Done()
|
||||
}
|
||||
out := make(chan delivery.Step)
|
||||
go func() {
|
||||
defer close(out)
|
||||
defer cancel()
|
||||
defer func() { stopBound() }()
|
||||
send := func(s delivery.Step) bool {
|
||||
select {
|
||||
case out <- s:
|
||||
return true
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
}
|
||||
}
|
||||
ended := func(reason string) delivery.Step {
|
||||
at := w.now().UTC()
|
||||
return delivery.Step{ID: askID, Node: LostNode, State: delivery.StepFailed, Reason: reason, Ended: at}
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case s, ok := <-inner:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if notTaken != nil {
|
||||
stopBound() // a node took it: its step is bounded by the asker from its start
|
||||
notTaken = nil
|
||||
}
|
||||
if !send(s) {
|
||||
return
|
||||
}
|
||||
case g := <-dead:
|
||||
dead = nil
|
||||
send(ended(g.Words()))
|
||||
return
|
||||
case <-notTaken:
|
||||
send(ended(fmt.Sprintf("%s within %s of its put; it was lost or waits behind a backlog; ask again", NotTakenWords, bound)))
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// IsLost says whether a step was ended by LostAskWaiter because the build seat gave its ask up (LostWords).
|
||||
func IsLost(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, LostWords) }
|
||||
|
||||
// IsNotTaken says whether a step was ended by LostAskWaiter because no build node took its ask within the bound.
|
||||
func IsNotTaken(s delivery.Step) bool {
|
||||
return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords)
|
||||
}
|
||||
|
||||
// **The asker hears DEAD_LETTERS, tries again when it cannot, and says whether it hears** (issue 734). The read is
|
||||
// granted with the asker's seat traffic (buscompose DeadLetterReads); an asker started before its grant was composed is
|
||||
// refused, and an asker that only tried once would stay deaf until it starts again, silently. DeadLetterHearing tries in
|
||||
// the background, never holding the start up, and again on an event (Again): the asker's reconnect to the bus, and every
|
||||
// ask waited for while it is deaf (LostAskWaiter.Hearing); no timer. Each change of whether it hears is said (Said), so
|
||||
// the module's health tool says it: a deaf asker is not healthy.
|
||||
|
||||
// Hearing is what a DeadLetterHearing says: whether it hears every build seat stream it follows, and why not.
|
||||
type Hearing struct {
|
||||
Heard bool `json:"heard"`
|
||||
Streams []string `json:"streams"`
|
||||
Why string `json:"why,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// HearingKey is the key a module keeps its Hearing under, in its own state (the checker's and the builder's hearing).
|
||||
const HearingKey = "dead-letters"
|
||||
|
||||
// DeadLetterFollower starts following one seat stream's copies in DEAD_LETTERS from since, handing each given-up ask to
|
||||
// heard; it answers the stop, or why it cannot.
|
||||
type DeadLetterFollower interface {
|
||||
Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error)
|
||||
}
|
||||
|
||||
// JetStreamDeadLetters is the bus side of a DeadLetterFollower: an ordered consumer on DEAD_LETTERS filtered to the
|
||||
// stream's copies, which resumes where it was after a reconnect and, reset before it heard any, starts at since again.
|
||||
type JetStreamDeadLetters struct{ JS jetstream.JetStream }
|
||||
|
||||
// Follow is DeadLetterFollower's.
|
||||
func (j JetStreamDeadLetters) Follow(ctx context.Context, stream string, since time.Time, heard func(GivenUp)) (func(), error) {
|
||||
cons, err := j.JS.OrderedConsumer(ctx, DeadLettersStream, jetstream.OrderedConsumerConfig{
|
||||
FilterSubjects: []string{DeadLetterFilter(stream)}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cc, err := cons.Consume(func(m jetstream.Msg) {
|
||||
if g, ok := GivenUpOf(m.Headers()); ok {
|
||||
heard(g)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return cc.Stop, nil
|
||||
}
|
||||
|
||||
// DeadLetterHearing follows the build seats' given-up asks into Dead. Make it with NewDeadLetterHearing.
|
||||
type DeadLetterHearing struct {
|
||||
follower DeadLetterFollower
|
||||
streams []string
|
||||
dead *DeadLetters
|
||||
logf func(string, ...any)
|
||||
said func(Hearing)
|
||||
ctx context.Context
|
||||
since time.Time
|
||||
|
||||
mu sync.Mutex
|
||||
stops map[string]func()
|
||||
why map[string]string
|
||||
trying bool
|
||||
again bool
|
||||
told *Hearing
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// NewDeadLetterHearing starts hearing the given-up asks of each language's build seat stream, in the background. said
|
||||
// is told each change of whether it hears (nil tells nothing); ctx ends it.
|
||||
func NewDeadLetterHearing(ctx context.Context, follower DeadLetterFollower, languages []string, dead *DeadLetters,
|
||||
logf func(string, ...any), said func(Hearing)) *DeadLetterHearing {
|
||||
h := &DeadLetterHearing{follower: follower, dead: dead, logf: logf, said: said, ctx: ctx, since: time.Now(),
|
||||
stops: map[string]func(){}, why: map[string]string{}}
|
||||
for _, l := range languages {
|
||||
s := SeatStream(l)
|
||||
h.streams = append(h.streams, s)
|
||||
h.why[s] = "not tried yet"
|
||||
}
|
||||
if h.logf == nil {
|
||||
h.logf = func(string, ...any) {}
|
||||
}
|
||||
h.Again()
|
||||
return h
|
||||
}
|
||||
|
||||
// Deaf says why a stream is not heard, or "" when every one is.
|
||||
func (h *DeadLetterHearing) Deaf() string {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return h.deafLocked()
|
||||
}
|
||||
|
||||
func (h *DeadLetterHearing) deafLocked() string {
|
||||
var out []string
|
||||
for _, s := range h.streams {
|
||||
if why, deaf := h.why[s]; deaf {
|
||||
out = append(out, s+": "+why)
|
||||
}
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return ""
|
||||
}
|
||||
return DeadLettersStream + " is not heard, so an ask a build seat gives up on is waited for until its bound: " +
|
||||
strings.Join(out, "; ")
|
||||
}
|
||||
|
||||
// Again tries, in the background, every stream not heard yet; a call while it tries makes it try once more after.
|
||||
func (h *DeadLetterHearing) Again() {
|
||||
h.mu.Lock()
|
||||
if h.trying {
|
||||
h.again = true
|
||||
h.mu.Unlock()
|
||||
return
|
||||
}
|
||||
h.trying = true
|
||||
h.mu.Unlock()
|
||||
h.wg.Add(1)
|
||||
go func() {
|
||||
defer h.wg.Done()
|
||||
for {
|
||||
h.try()
|
||||
h.mu.Lock()
|
||||
if !h.again || h.ctx.Err() != nil {
|
||||
h.trying = false
|
||||
h.mu.Unlock()
|
||||
return
|
||||
}
|
||||
h.again = false
|
||||
h.mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Wait returns when no try runs.
|
||||
func (h *DeadLetterHearing) Wait() { h.wg.Wait() }
|
||||
|
||||
// Stop ends every stream's following.
|
||||
func (h *DeadLetterHearing) Stop() {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
for s, stop := range h.stops {
|
||||
stop()
|
||||
delete(h.stops, s)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *DeadLetterHearing) try() {
|
||||
for _, s := range h.streams {
|
||||
h.mu.Lock()
|
||||
_, deaf := h.why[s]
|
||||
h.mu.Unlock()
|
||||
if !deaf || h.ctx.Err() != nil {
|
||||
continue
|
||||
}
|
||||
stream := s
|
||||
stop, err := h.follower.Follow(h.ctx, stream, h.since, func(g GivenUp) {
|
||||
h.logf("the ask %s on %s was given up after %d deliveries at %s: the wait for its step ends", g.AskID, g.Stream,
|
||||
g.Deliveries, g.At.Format(time.RFC3339))
|
||||
h.dead.Heard(g)
|
||||
})
|
||||
h.mu.Lock()
|
||||
if err != nil {
|
||||
h.why[stream] = err.Error()
|
||||
h.logf("the asks %s gives up on are not heard (%s of %s); tried again at the next reconnect or ask: %v",
|
||||
stream, DeadLetterFilter(stream), DeadLettersStream, err)
|
||||
} else {
|
||||
delete(h.why, stream)
|
||||
h.stops[stream] = stop
|
||||
h.logf("the asks %s gives up on are heard from %s", stream, DeadLettersStream)
|
||||
}
|
||||
h.mu.Unlock()
|
||||
}
|
||||
h.tell()
|
||||
}
|
||||
|
||||
// tell says the hearing when it changed.
|
||||
func (h *DeadLetterHearing) tell() {
|
||||
h.mu.Lock()
|
||||
why := h.deafLocked()
|
||||
now := Hearing{Heard: why == "", Streams: append([]string(nil), h.streams...), Why: why, At: time.Now().UTC()}
|
||||
same := h.told != nil && h.told.Heard == now.Heard && h.told.Why == now.Why
|
||||
if !same {
|
||||
h.told = &now
|
||||
}
|
||||
h.mu.Unlock()
|
||||
if !same && h.said != nil {
|
||||
h.said(now)
|
||||
}
|
||||
}
|
||||
|
||||
// HearingState is the name of the state an asker keeps its Hearing in (`<module>_hearing`, key HearingKey), which its
|
||||
// health tool reads.
|
||||
const HearingState = "hearing"
|
||||
|
||||
// SayHearing is a DeadLetterHearing's said that keeps each Hearing in kv under HearingKey; nil kv keeps nothing and
|
||||
// says so once.
|
||||
func SayHearing(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) func(Hearing) {
|
||||
if kv == nil {
|
||||
if logf != nil {
|
||||
logf("whether %s is heard is not kept: the state %s is not there, so the health tool cannot say it", DeadLettersStream, HearingState)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return func(h Hearing) {
|
||||
data, err := json.Marshal(h)
|
||||
if err == nil {
|
||||
_, err = kv.Put(ctx, HearingKey, data)
|
||||
}
|
||||
if err != nil && logf != nil {
|
||||
logf("whether %s is heard could not be kept: %v", DeadLettersStream, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// HearingHealth is the health answer of an asker's kept Hearing (raw, empty when none is kept): healthy while it hears
|
||||
// DEAD_LETTERS, unhealthy saying why while it does not or has not said.
|
||||
func HearingHealth(raw []byte) health.Answer {
|
||||
if len(raw) == 0 {
|
||||
return health.Unhealthy("it has not said whether it hears " + DeadLettersStream + " (issue 734): it is starting, or its state " + HearingState + " is not there")
|
||||
}
|
||||
var h Hearing
|
||||
if err := json.Unmarshal(raw, &h); err != nil {
|
||||
return health.Unhealthy("what it keeps of hearing " + DeadLettersStream + " is not a hearing: " + err.Error())
|
||||
}
|
||||
if !h.Heard {
|
||||
return health.Unhealthy(h.Why + " (since " + h.At.Format(time.RFC3339) + "; tried again at its next reconnect or ask)")
|
||||
}
|
||||
return health.Healthy("it hears " + DeadLettersStream + " for " + strings.Join(h.Streams, ", "))
|
||||
}
|
||||
Reference in New Issue
Block a user