Checker and builder: wait on the build seats through mesh-sdk's LostAskWaiter, so an ask the seat gave up on ends every asker's wait (issue 734)
The waiter moved to mesh-sdk buildrun (fix/734-lost-ask, d96b9fd6), where every asker of the build seats shares it. The builder hears DEAD_LETTERS for the go, node and image seats and asks a given-up part of a build again at once, with the seat's words as why, instead of waiting for the check bound. The checker keeps what it does with a lost ask (a try ends as error, a check is asked again) and its verdict words.
This commit is contained in:
@@ -149,6 +149,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
|
||||
// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
|
||||
asks := buildrun.JetStreamAsks{JS: js}
|
||||
// The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the wait for its step at once,
|
||||
// and the lane asks its part again (Builder.waitOnce) instead of waiting for the check bound.
|
||||
dead := &buildrun.DeadLetters{}
|
||||
defer buildrun.HearDeadLetters(ctx, js, delivery.AskLanguages, dead, logger.Printf)()
|
||||
// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
|
||||
// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
|
||||
account, err := process.ReadAccount(path)
|
||||
@@ -162,7 +166,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
defer calls.Close()
|
||||
b := &build.Builder{
|
||||
Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
|
||||
Asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
|
||||
Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: asks, Dead: dead}, Bound: bound},
|
||||
Held: heldOnTheBus{js},
|
||||
// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
|
||||
Gone: heldOnTheBus{js},
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-builder
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130046-d96b9fd66069
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 h1:fzOkrz9CSp3mEjeV7NfZx8yoivI1sD1AOtIPtyyF8RI=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130046-d96b9fd66069 h1:dPwZwGbW8LuT4p2baXuiyVNOWU6XO46kbw46VU5TAUQ=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026130046-d96b9fd66069/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
// **An ask that can no longer be answered is asked again** (novox/hq issue 1945). A build seat's stream gives an ask up
|
||||
@@ -20,7 +21,8 @@ import (
|
||||
// restart take over the same id and wait again. The builder notices it at the check bound (CheckAfter, the step bound),
|
||||
// a context's bound and no timer of its own: it asks Gone, which reads the bus, and asks such an ask again under a new
|
||||
// id, written into the key, said in its log and as the event build.asked-again, at most Askings times; then the key is
|
||||
// failed with why. A person asks the same by hand with the tool builder_ask_again (AskAgain).
|
||||
// failed with why. An ask DEAD_LETTERS says the seat gave up (buildrun.LostAskWaiter, issue 734) is asked again at once.
|
||||
// A person asks the same by hand with the tool builder_ask_again (AskAgain).
|
||||
|
||||
// DefaultAskings is how many times a build module's part of a build is asked, the first included, when Builder.Askings is 0.
|
||||
const DefaultAskings = 3
|
||||
@@ -315,6 +317,11 @@ func (b *Builder) waitOnce(ctx context.Context, ask delivery.StepAsk, again <-ch
|
||||
defer stop()
|
||||
}
|
||||
step, err := b.Asks.Wait(wait, ask)
|
||||
if err == nil && buildrun.IsLost(step) {
|
||||
// The build seat gave the ask up (issue 734): it can no longer be answered, so it is asked again now, with the
|
||||
// words of why, not at the check bound.
|
||||
return step, step.Reason, errAgain
|
||||
}
|
||||
if err == nil || ctx.Err() != nil {
|
||||
return step, "", err
|
||||
}
|
||||
|
||||
@@ -690,12 +690,14 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
|
||||
step string
|
||||
byHand bool
|
||||
always bool // Gone says every ask cannot be answered, not only the first
|
||||
dead bool // DEAD_LETTERS keeps the first ask: the seat gave it up (issue 734)
|
||||
puts int
|
||||
done bool
|
||||
asking int // the asking the event says, 0 when none is said
|
||||
}{
|
||||
{name: "given up by its stream: asked again under a new id and done from it", puts: 2, done: true, asking: 2},
|
||||
{name: "asked again by hand: the same, with no check bound", byHand: true, puts: 2, done: true, asking: 2},
|
||||
{name: "given up into DEAD_LETTERS: its wait ends failed with the words and it is asked again at once", dead: true, puts: 2, done: true, asking: 2},
|
||||
{name: "never answerable: failed after its askings", askings: 2, always: true, puts: 2},
|
||||
{name: "taken over at its last asking: failed, not asked again", askings: 3, always: true, step: "asked mesh-build-go again (asking 3): x", puts: 0},
|
||||
} {
|
||||
@@ -714,7 +716,11 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
|
||||
r.st.vals[r.key] = data
|
||||
}
|
||||
gone := &goneAsks{}
|
||||
if !c.byHand {
|
||||
dead := &buildrun.DeadLetters{}
|
||||
if c.dead {
|
||||
r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: buildrun.LostAskWaiter{Inner: r.asks, Dead: dead}, Bound: 2 * time.Second}
|
||||
}
|
||||
if !c.byHand && !c.dead {
|
||||
gone.why = func(a delivery.StepAsk) string {
|
||||
if c.always || a.ID == first || first == "" {
|
||||
return "it waits first in line while a node took a later ask"
|
||||
@@ -724,6 +730,10 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
|
||||
r.b.Gone, r.b.CheckAfter = gone, 5*time.Millisecond
|
||||
}
|
||||
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
|
||||
if c.dead {
|
||||
dead.Heard(buildrun.GivenUp{AskID: waitForAsks(t, r, 1)[0].ID, Stream: "SEAT_NODE_BUILD_GO", Deliveries: 5,
|
||||
At: time.Date(2026, 10, 26, 12, 26, 40, 0, time.UTC)})
|
||||
}
|
||||
if c.byHand {
|
||||
waitForAsks(t, r, 1)
|
||||
for deadline := time.Now().Add(2 * time.Second); len(r.b.AskAgain(r.key, "stuck since 19:06")) == 0; time.Sleep(time.Millisecond) {
|
||||
@@ -757,13 +767,16 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
// Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream.
|
||||
if gone.mu.Lock(); c.done && !c.byHand && (len(gone.dropped) != 1 || gone.dropped[0] != first) {
|
||||
if gone.mu.Lock(); c.done && !c.byHand && !c.dead && (len(gone.dropped) != 1 || gone.dropped[0] != first) {
|
||||
t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first)
|
||||
}
|
||||
gone.mu.Unlock()
|
||||
if c.asking > 0 && (len(said) != 1 || said[0].Asking != c.asking || said[0].New == said[0].Old || said[0].Key != r.key || said[0].Why == "") {
|
||||
t.Errorf("said %+v, want once at asking %d", said, c.asking)
|
||||
}
|
||||
if c.dead && (len(said) != 1 || !strings.Contains(said[0].Why, buildrun.LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z")) {
|
||||
t.Errorf("said %+v, want the ask asked again saying the seat gave it up", said)
|
||||
}
|
||||
})
|
||||
}
|
||||
t.Run("what the bus says of an ask", func(t *testing.T) { // 1945
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
@@ -304,7 +305,7 @@ func startBed(t *testing.T) *bed {
|
||||
b.tools[module] = kv
|
||||
}
|
||||
// What the controller keeps of the messages a consumer gave up on (issue 734).
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: check.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}}); err != nil {
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buildrun.DeadLettersStream, Subjects: []string{"mesh.events.dead.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The checker's stream of events, which the indexer and the forge bridge hear.
|
||||
@@ -749,10 +750,10 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
|
||||
t.Fatal("the held try's ask was never taken")
|
||||
}
|
||||
dead := &nats.Msg{Subject: "mesh.events.dead.SEAT_NODE_BUILD_GO.SEAT_NODE_BUILD_GO_CHECK_worker", Header: nats.Header{}}
|
||||
dead.Header.Set(check.DeadHeaderMsgID, askID)
|
||||
dead.Header.Set(check.DeadHeaderDeliveries, "5")
|
||||
dead.Header.Set(check.DeadHeaderStream, "SEAT_NODE_BUILD_GO")
|
||||
dead.Header.Set(check.DeadHeaderGaveUp, time.Now().UTC().Format(time.RFC3339Nano))
|
||||
dead.Header.Set(buildrun.DeadHeaderMsgID, askID)
|
||||
dead.Header.Set(buildrun.DeadHeaderDeliveries, "5")
|
||||
dead.Header.Set(buildrun.DeadHeaderStream, "SEAT_NODE_BUILD_GO")
|
||||
dead.Header.Set(buildrun.DeadHeaderGaveUp, time.Now().UTC().Format(time.RFC3339Nano))
|
||||
if _, err := b.js.PublishMsg(b.ctx, dead); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -762,7 +763,7 @@ func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
|
||||
break
|
||||
}
|
||||
}
|
||||
if last.State != check.TryError || !strings.Contains(last.Said, check.LostWords+" after 5 deliveries") {
|
||||
if last.State != check.TryError || !strings.Contains(last.Said, buildrun.LostWords+" after 5 deliveries") {
|
||||
t.Fatalf("the try whose ask was given up = %+v, want error saying so", last)
|
||||
}
|
||||
if again, err := check.AskTry(store, map[string]string{repo: trunk}, check.TryAsk{Repository: repo, Head: head2}, time.Now()); err != nil || again == lost {
|
||||
|
||||
@@ -199,9 +199,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step.
|
||||
asks := buildrun.JetStreamAsks{JS: js}
|
||||
// The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the step its asker waits for.
|
||||
dead := &check.DeadLetters{}
|
||||
defer hearDeadLetters(ctx, js, dead, logger.Printf)()
|
||||
waiter := check.LostAskWaiter{Inner: asks, Dead: dead}
|
||||
dead := &buildrun.DeadLetters{}
|
||||
defer buildrun.HearDeadLetters(ctx, js, []string{delivery.AskGo, delivery.AskNode}, dead, logger.Printf)()
|
||||
waiter := buildrun.LostAskWaiter{Inner: asks, Dead: dead}
|
||||
ch := &check.Checker{
|
||||
Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]},
|
||||
Owed: kvStore{buckets[delivery.CheckerOwedBucket]},
|
||||
@@ -401,43 +401,3 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// hearDeadLetters follows, for each build seat's stream, the copies DEAD_LETTERS keeps of the asks its workers gave up on
|
||||
// (issue 734): an ordered consumer from this start, which resumes where it was after a reconnect, so a bus reload loses none. Its
|
||||
// grant is a user's of a seat pulled by kind (mesh-sdk buscompose DeadLetterReads); where the bus refuses it, the checker
|
||||
// says so and goes on: a try still ends when no node took its ask within the step bound, and a check waits as before.
|
||||
func hearDeadLetters(ctx context.Context, js jetstream.JetStream, dead *check.DeadLetters, logf func(string, ...any)) func() {
|
||||
var stops []func()
|
||||
// From this start on: a step this process waits for was asked since, and a reset of the consumer before it heard any
|
||||
// copy starts here again rather than at the reset.
|
||||
since := time.Now()
|
||||
for _, stream := range check.BuildSeatStreams {
|
||||
filter := check.DeadLetterFilter(stream)
|
||||
notHeard := func(err error) {
|
||||
logf("the asks %s gives up on are not heard (%s of %s), so a check whose ask it gives up waits until the checker starts again: %v",
|
||||
stream, filter, check.DeadLettersStream, err)
|
||||
}
|
||||
cons, err := js.OrderedConsumer(ctx, check.DeadLettersStream, jetstream.OrderedConsumerConfig{
|
||||
FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverByStartTimePolicy, OptStartTime: &since})
|
||||
if err != nil {
|
||||
notHeard(err)
|
||||
continue
|
||||
}
|
||||
cc, err := cons.Consume(func(m jetstream.Msg) {
|
||||
if g, ok := check.GivenUpOf(m.Headers()); ok {
|
||||
logf("the ask %s on %s was given up after %d deliveries at %s: the step that waits for it ends", g.AskID, g.Stream, g.Deliveries, g.At.Format(time.RFC3339))
|
||||
dead.Heard(g)
|
||||
}
|
||||
}, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logf("hearing %s: %v", check.DeadLettersStream, err) }))
|
||||
if err != nil {
|
||||
notHeard(err)
|
||||
continue
|
||||
}
|
||||
stops = append(stops, cc.Stop)
|
||||
}
|
||||
return func() {
|
||||
for _, stop := range stops {
|
||||
stop()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,257 +1,30 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
// **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.
|
||||
//
|
||||
// The checker hears those copies (DeadLetters, an ordered consumer on DEAD_LETTERS filtered to the build seats' streams,
|
||||
// granted to a user of a seat pulled by kind, mesh-sdk buscompose DeadLetterReads) and LostAskWaiter ends the waiting step
|
||||
// at once as failed, its words saying LostWords. A try ends as error; a check asks again (Checker.try).
|
||||
//
|
||||
// Belt and braces, for a try only (NotTakenBound): an ask no build node took within the step bound ends the same way,
|
||||
// saying NotTakenWords. A check's ask waits in the stream as long as the backlog is (ADR 0525 §9): a backlog is no fault,
|
||||
// and a check ended for one would owe its pull request a failure.
|
||||
|
||||
// The words of a step the checker ended for its ask.
|
||||
const (
|
||||
LostWords = "its ask was given up by the build seat"
|
||||
NotTakenWords = "no build node took its ask"
|
||||
)
|
||||
// **An ask the build seat gave up on ends its step** (novox/hq issue 734): mesh-sdk buildrun.LostAskWaiter hands the
|
||||
// waiting asker a failed step on the node buildrun.LostNode when DEAD_LETTERS keeps the ask's copy (buildrun.HearDeadLetters,
|
||||
// granted by buscompose DeadLetterReads), and, under buildrun.NotTakenBound, when no build node took the ask within the
|
||||
// bound. What follows is the checker's: a try ends as error (Tryer.run), with the not-taken bound at the step bound
|
||||
// (Tryer.NotTaken); a check is asked again under its claim, at most LostAskTries times, then fails as not heard and is
|
||||
// judged again when the nodes change (Checker.try). A check's ask has no not-taken bound: it waits in the stream as long as
|
||||
// the backlog is (ADR 0525 §9), and a check ended for a backlog would owe its pull request a failure.
|
||||
|
||||
// LostAskTries is how often a check asks again whose ask was given up, before it fails as not heard (UnheardWords) and is
|
||||
// judged again when the nodes change.
|
||||
const LostAskTries = 3
|
||||
|
||||
// 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 (mesh-sdk 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 + ".>" }
|
||||
|
||||
// BuildSeatStreams are the build seats' streams the checker asks on (mesh-sdk buscompose seatStreamName of node-build-go
|
||||
// and node-build-node).
|
||||
var BuildSeatStreams = []string{"SEAT_NODE_BUILD_GO", "SEAT_NODE_BUILD_NODE"}
|
||||
|
||||
// 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 a buildrun.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 buildrun.AskWaiter
|
||||
Dead *DeadLetters
|
||||
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) {
|
||||
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: "bus", 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
|
||||
}
|
||||
|
||||
// IsLostAsk says whether a verdict failed because the build seat gave its ask up (LostWords).
|
||||
// IsLostAsk says whether a verdict failed because the build seat gave its ask up (buildrun.LostWords).
|
||||
func IsLostAsk(v delivery.Verdict) bool {
|
||||
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, LostWords)
|
||||
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.LostWords)
|
||||
}
|
||||
|
||||
// IsNotTaken says whether a verdict failed because no build node took its ask within the bound (NotTakenWords).
|
||||
// IsNotTaken says whether a verdict failed because no build node took its ask within the bound (buildrun.NotTakenWords).
|
||||
func IsNotTaken(v delivery.Verdict) bool {
|
||||
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, NotTakenWords)
|
||||
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.NotTakenWords)
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ import (
|
||||
type losingAsks struct {
|
||||
*buildrun.FakeAsks
|
||||
answer func(delivery.StepAsk) delivery.Step
|
||||
dead *DeadLetters
|
||||
dead *buildrun.DeadLetters
|
||||
never bool
|
||||
|
||||
mu sync.Mutex
|
||||
@@ -38,9 +38,9 @@ func (l *losingAsks) PublishAsk(ctx context.Context, subject, id string, data []
|
||||
switch {
|
||||
case l.never:
|
||||
case lost:
|
||||
h := map[string][]string{DeadHeaderMsgID: {id}, DeadHeaderDeliveries: {"5"}, DeadHeaderStream: {"SEAT_NODE_BUILD_GO"},
|
||||
DeadHeaderGaveUp: {"2026-10-26T12:26:40Z"}}
|
||||
g, ok := GivenUpOf(h)
|
||||
h := map[string][]string{buildrun.DeadHeaderMsgID: {id}, buildrun.DeadHeaderDeliveries: {"5"}, buildrun.DeadHeaderStream: {"SEAT_NODE_BUILD_GO"},
|
||||
buildrun.DeadHeaderGaveUp: {"2026-10-26T12:26:40Z"}}
|
||||
g, ok := buildrun.GivenUpOf(h)
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
@@ -56,8 +56,8 @@ func (l *losingAsks) PublishAsk(ctx context.Context, subject, id string, data []
|
||||
}
|
||||
|
||||
func losing(b *bench, lose int) *losingAsks {
|
||||
l := &losingAsks{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, dead: &DeadLetters{}, lose: lose}
|
||||
b.ex.Asks = buildrun.Asker{Publisher: l, Waiter: LostAskWaiter{Inner: l.FakeAsks, Dead: l.dead}, Bound: 5 * time.Second}
|
||||
l := &losingAsks{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, dead: &buildrun.DeadLetters{}, lose: lose}
|
||||
b.ex.Asks = buildrun.Asker{Publisher: l, Waiter: buildrun.LostAskWaiter{Inner: l.FakeAsks, Dead: l.dead}, Bound: 5 * time.Second}
|
||||
return l
|
||||
}
|
||||
|
||||
@@ -83,8 +83,8 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi
|
||||
never bool
|
||||
want string
|
||||
}{
|
||||
{"given up by the build seat", 1, false, LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"},
|
||||
{"taken by no build node within the bound", 0, true, NotTakenWords + " within 200ms"},
|
||||
{"given up by the build seat", 1, false, buildrun.LostWords + " after 5 deliveries at 2026-10-26T12:26:40Z"},
|
||||
{"taken by no build node within the bound", 0, true, buildrun.NotTakenWords + " within 200ms"},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
kv := newMemKV()
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
// A try is a pushed head judged by the checker as a pull request's head is, before its pull request is opened (ADR 0501
|
||||
@@ -399,7 +400,7 @@ func (tr *Tryer) run(ctx context.Context, key string) {
|
||||
Try: &TryOptions{Level: t.Level, Run: t.Tests}}
|
||||
judging := ctx
|
||||
if tr.NotTaken > 0 {
|
||||
judging = NotTakenBound(ctx, tr.NotTaken)
|
||||
judging = buildrun.NotTakenBound(ctx, tr.NotTaken)
|
||||
}
|
||||
out, judged := judgeWithin(judging, tr.Judge, cs, tr.Bound)
|
||||
if ctx.Err() != nil {
|
||||
|
||||
Reference in New Issue
Block a user