diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index 5e88371c..d9a0c35b 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -14,6 +14,7 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-delivery/sdk/buildstep" + "git.novox.be/novox/mesh-sdk/go/busretry" ) // **An asker puts an ask on its build seat's stream and waits for its step** (novox/hq ADR 0525 §1, §6; issue 1887 A4, @@ -98,7 +99,14 @@ func (k Asker) dropped(a buildstep.StepAsk, why string) { } } -// Put puts a on its stream: the seat's accept subject of its language and kind. +// PutTry is how long one publish of an ask waits for its stream's answer before it is made again (Put). +var PutTry = 10 * time.Second + +// Put puts a on its stream: the seat's accept subject of its language and kind. The put is a kept call (novox/hq ADR 0668 +// §1, §2; issue 1966, issue 1748 note 27): it carries the ask's id as its message id, so the stream takes it once, and it +// is made again through every transport fault (a deadline, no responders) within busretry.Bound, each try bounded by +// PutTry, until ctx ends. A put that still meets one is an error opening with buildstep.TransportFaultWords, so the key it +// fails is one the builder does not take again by itself (ADR 0664 §3) and builder_build_again does. func (k Asker) Put(ctx context.Context, a buildstep.StepAsk) error { raw, err := buildstep.EncodeStepAsk(a) if err != nil { @@ -108,10 +116,25 @@ func (k Asker) Put(ctx context.Context, a buildstep.StepAsk) error { if err != nil { return err } - if err := k.Publisher.PublishAsk(ctx, subject, a.ID, raw); err != nil { - return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err) + _, err = busretry.DoRepeatable("putting the "+a.Kind+" ask "+a.ID, func() (struct{}, error) { + if err := ctx.Err(); err != nil { + return struct{}{}, fmt.Errorf("the put was given up: %v", err) // not a fault to wait out: the build ended + } + try, cancel := context.WithTimeout(ctx, PutTry) + defer cancel() + err := k.Publisher.PublishAsk(try, subject, a.ID, raw) + if err != nil && ctx.Err() != nil { + return struct{}{}, fmt.Errorf("the put was given up: %v", err) + } + return struct{}{}, err + }, nil) + switch { + case err == nil: + return nil + case busretry.TimedOut(err) || busretry.Undelivered(err): + return fmt.Errorf("%sthe put of the %s ask %s: %w", buildstep.TransportFaultWords, a.Kind, a.ID, err) } - return nil + return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err) } // Wait waits for the step of an ask already put (a builder that started again takes over its own asks with it, never diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index 29406198..a460f924 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -5,12 +5,16 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strings" "sync" "testing" "time" + "github.com/nats-io/nats.go" + "git.novox.be/novox/mesh-delivery/sdk/buildstep" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -41,6 +45,39 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") { t.Fatalf("a refused step: %v", err) } + // Issue 1966 (ADR 0668 §1, §2; 1748 note 27): the put is a kept call under the ask's id, made again through a deadline and + // no responders until the stream takes it; a refusal of the stream is not made again; a put that meets a transport fault + // past the bound says it in buildstep.TransportFaultWords, so the key it fails is the hand tool's to build again. + t.Cleanup(busretry.Waits(time.Second, time.Millisecond, time.Millisecond)) + for name, c := range map[string]struct { + faults []error + tries int + transport bool + }{ + "a deadline, then no responders, then taken": {faults: []error{context.DeadlineExceeded, nats.ErrNoResponders}, tries: 3}, + "refused by the stream": {faults: []error{errors.New("maximum messages exceeded")}, tries: 1}, + "a deadline past the bound": {faults: slices.Repeat([]error{nats.ErrTimeout}, 10000), transport: true}, + } { + p := &faultyPuts{faults: c.faults, inner: &FakeAsks{}} + restore := func() {} + if c.transport { + restore = busretry.Waits(20*time.Millisecond, time.Millisecond, time.Millisecond) // a bound this row passes + } + err := (Asker{Publisher: p, Waiter: p.inner}).Put(context.Background(), goAsk("mvl0frv18d1960")) + restore() + switch { + case c.transport: + if err == nil || !strings.HasPrefix(err.Error(), buildstep.TransportFaultWords) { + t.Errorf("%s: %v, want it opens with the transport fault's words", name, err) + } + case c.tries == 1: + if err == nil || p.tries != 1 || strings.Contains(err.Error(), buildstep.TransportFaultWords) { + t.Errorf("%s: %v after %d tries, want one try and no transport fault", name, err, p.tries) + } + case err != nil || p.tries != c.tries || len(p.inner.Asks()) != 1 || len(p.ids) != c.tries || p.ids[0] != p.ids[len(p.ids)-1]: + t.Errorf("%s: %v after %d tries under %v, %d asks taken, want %d tries under one id, one ask", name, err, p.tries, p.ids, len(p.inner.Asks()), c.tries) + } + } // Issue 1748 (ADR 0664 §3): a run that ended on a transport fault is no end; the next run's end is the answer, in either // arrival order; three runs then failed say a transport fault on 3 runs with each node; a bus that never gives the ask // again ends the wait saying so. @@ -320,3 +357,27 @@ type failingDrops struct{} func (failingDrops) DropAsk(context.Context, buildstep.StepAsk, string) error { return errors.New("no holder answered") } + +// faultyPuts fails its first puts with faults, then puts through inner (issue 1966). +type faultyPuts struct { + mu sync.Mutex + faults []error + tries int + ids []string + inner *FakeAsks +} + +func (p *faultyPuts) PublishAsk(ctx context.Context, subject, id string, data []byte) error { + p.mu.Lock() + p.tries++ + p.ids = append(p.ids, id) + var err error + if len(p.faults) > 0 { + err, p.faults = p.faults[0], p.faults[1:] + } + p.mu.Unlock() + if err != nil { + return err + } + return p.inner.PublishAsk(ctx, subject, id, data) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index cc24d6ec..002ecd3d 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -160,10 +160,11 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q {Name: build.BuildAgainTool, Description: "Build a failed key again (rebuild a module, build a failed key again): the builder writes the key of module and " + "fingerprint building as a new revision of the same key and asks the build seats, as it does by itself when it looks at a module " + - "again (ADR 0429 clause 18). Answers the new revision and the ask id of each build module (go, node, image). Refuses a key that " + - "is not failed, and says why: done (nothing to do), or waiting (building or queued: " + build.AskAgainTool + " asks its waiting asks " + - "again); a key failed for a reason the builder never retries on the same fingerprint (a pin missing, a version taken, digests that " + - "differ) and a fingerprint its module no longer has are refused too. Use it for a key builder_stuck does not show, which failed.", + "again (ADR 0429 clause 18). Answers the new revision and the ask id of each build module (go, node, image). Also takes a key left " + + "building that no look of the builder holds (its ask never put or its end never written, ADR 0668 §5): an ask the key names and " + + "its stream holds is waited for, one never put is put under its id. Refuses a key done (nothing to do) or queued (" + + build.AskAgainTool + " asks its waiting asks again), and one the builder is looking at now; a key failed for a reason the builder never retries on the same fingerprint (a pin missing, a version taken, digests that " + + "differ) and a fingerprint its module no longer has are refused too. Use it for a failed key, or one builder_stuck shows building with no ask or no step.", Input: map[string]any{"type": "object", "required": []string{"module", "fp", "why"}, "properties": map[string]any{ "module": map[string]any{"type": "string", "description": "the module's name"}, "fp": map[string]any{"type": "string", "description": "its new-form fingerprint, src2:"}, diff --git a/modules/mesh-delivery-builder/internal/build/buildagain.go b/modules/mesh-delivery-builder/internal/build/buildagain.go index 6f39187e..2c135769 100644 --- a/modules/mesh-delivery-builder/internal/build/buildagain.go +++ b/modules/mesh-delivery-builder/internal/build/buildagain.go @@ -84,7 +84,8 @@ func (b *Builder) handEnded(key string) { // BuildAgain builds the failed key of module at fp again, as the builder does by itself: the key is written building as a // new revision and the build module asks are put on the build seats. It answers once the asks are put, or at bound. A key -// that is not failed is refused with why: done (nothing to do) or waiting (builder_ask_again asks its waiting asks again). +// left building that no look of this builder holds is taken as it stands and its asks put if they never were (issue 1966). +// A key done or queued is refused with why: done (nothing to do) or waiting (builder_ask_again asks its waiting asks again). // ctx is the builder's own context: the build it starts lives on it. func (b *Builder) BuildAgain(ctx context.Context, module, fp, why, by string, bound time.Duration) BuildAgainResult { b.init() @@ -106,8 +107,12 @@ func (b *Builder) BuildAgain(ctx context.Context, module, fp, why, by string, bo switch { case cur.State == builds.BuildDone: return BuildAgainResult{Refused: fmt.Sprintf("%s is done: nothing to do", key)} - case cur.State == builds.BuildBuilding || cur.State == builds.BuildQueued: + case cur.State == builds.BuildQueued: return BuildAgainResult{Refused: fmt.Sprintf("%s is waiting (%s: %s): %s asks its waiting asks again", key, cur.State, cur.Step, AskAgainTool)} + case cur.State == builds.BuildBuilding: + // A key left building that no look of this builder holds (refused below when one does): its put or its end was never + // written (issue 1966, 1748 note 27: "building, claimed", no ask). It is taken as it stands, an ask it names and the + // stream holds is waited for, one never put is put under its id, one never named is asked (ADR 0668 §5). case cur.State != builds.BuildFailed: return BuildAgainResult{Refused: fmt.Sprintf("%s is %q, not failed", key, cur.State)} case builds.FailedUntilChanged(cur.Error) || strings.HasPrefix(cur.Error, buildrun.DigestsDifferPrefix): @@ -139,7 +144,11 @@ func (b *Builder) BuildAgain(ctx context.Context, module, fp, why, by string, bo } b.mu.Unlock() }() - b.logf("%s: failed (%s); built again by hand, asked by %s: %s", key, cur.Error, by, why) + if cur.State == builds.BuildBuilding { + b.logf("%s: left building (%s) with no look of this builder; built again by hand, asked by %s: %s", key, cur.Step, by, why) + } else { + b.logf("%s: failed (%s); built again by hand, asked by %s: %s", key, cur.Error, by, why) + } b.startOnce(ctx, key, module, m) wait, stop := context.WithTimeout(ctx, bound) // a context's bound for a person's answer (ADR 0437 §8) defer stop() diff --git a/modules/mesh-delivery-builder/internal/build/claim_test.go b/modules/mesh-delivery-builder/internal/build/claim_test.go index 04b68925..eeda081f 100644 --- a/modules/mesh-delivery-builder/internal/build/claim_test.go +++ b/modules/mesh-delivery-builder/internal/build/claim_test.go @@ -18,6 +18,8 @@ type memStore struct { writes int // before runs once ahead of the next write, so a test changes the key under the writer. before func() + // faults fail the next updates, one each, writing nothing (a deadline met on the bus). + faults []error } func newMem() *memStore { return &memStore{vals: map[string][]byte{}, revs: map[string]uint64{}} } @@ -67,6 +69,11 @@ func (m *memStore) Update(_ context.Context, key string, value []byte, rev uint6 m.mu.Lock() defer m.mu.Unlock() m.hook() + if len(m.faults) > 0 { + err := m.faults[0] + m.faults = m.faults[1:] + return 0, err + } if _, ok := m.vals[key]; !ok || m.revs[key] != rev { return 0, ErrConflict } diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 8b74d448..36a9b468 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -21,6 +21,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds/forms" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" repotrunk "git.novox.be/novox/mesh-delivery/sdk/trunk" + "git.novox.be/novox/mesh-sdk/go/busretry" "git.novox.be/novox/mesh-sdk/go/catalogue" "mesh-delivery-builder/internal/versions" @@ -477,7 +478,9 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB return } } - cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first, ByHand: byHand, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished, + // A key built again by hand is taken over as it stands too: one left building by a look that ended without its write + // (issue 1966) is the hand tool's to finish. + cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first || byHand, ByHand: byHand, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished, Stored: b.storedOnce(key)}) if err != nil { b.logf("%s: %v", module, err) @@ -924,7 +927,13 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m mes } } else { if err := asked.note(ctx, language, ask.ID, "asked "+l.module); err != nil { - l.err = fmt.Errorf("the ask of %s could not be written to the key: %w", l.module, err) + // The note and the put are one act (novox/hq ADR 0668 §1; issue 1748 note 27): a note a transport fault kept from + // the key fails the key with that fault's words, which builder_build_again takes (ADR 0664 §3). + fault := "" + if busretry.TimedOut(err) || busretry.Undelivered(err) { + fault = sdkstep.TransportFaultWords + "the note of the ask: " + } + l.err = fmt.Errorf("%sthe ask of %s could not be written to the key: %w", fault, l.module, err) return } if err := b.Asks.Put(ctx, ask); err != nil { diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index b4620e5b..a1429211 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -981,6 +981,8 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { failed string // during is a look of the builder's own at the module while the by-hand build runs: it still runs afterwards. during bool + // standing is a key taken as it stands (left building): its claim is the key's own revision, not a new one. + standing bool }{ {name: "build again: a look the builder queued during the by-hand build still runs afterwards", again: true, during: true, key: &builds.BuilderBuild{State: builds.BuildFailed, Error: "ace: no tun"}}, @@ -1002,8 +1004,16 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { key: &builds.BuilderBuild{State: builds.BuildFailed, Error: "ace: the step failed: no tun"}}, {name: "build again: a done key is refused, nothing to do", again: true, refuse: "is done: nothing to do", key: &builds.BuilderBuild{State: builds.BuildDone, Step: "built"}}, - {name: "build again: a building key is refused, builder_ask_again is for it", again: true, refuse: "is waiting", - key: &builds.BuilderBuild{State: builds.BuildBuilding, By: "novox.builder", Step: "asked mesh-build-go", Asks: map[string]string{buildstep.AskGo: "mvl0frv18d1960"}}}, + {name: "build again: a queued key is refused, builder_ask_again is for it", again: true, refuse: "is waiting", + key: &builds.BuilderBuild{State: builds.BuildQueued, By: "novox.builder", Step: "waits for the package x", Waits: "x@v1.0.0"}}, + // Issue 1966 (ADR 0668 §5; 1748 note 27): a key left building with no ask, no look of the builder holding it, is taken as + // it stands, by this builder or from another, asked and done; one a look holds is refused. + {name: "build again: a key left building claimed and unasked is asked and done", again: true, standing: true, + key: &builds.BuilderBuild{State: builds.BuildBuilding, By: "novox.builder", Step: "claimed"}}, + {name: "build again: a key another builder left building unasked is taken over", again: true, standing: true, + key: &builds.BuilderBuild{State: builds.BuildBuilding, By: "ace.builder", Step: "claimed"}}, + {name: "build again: a key left building that a look holds is refused", again: true, refuse: "is being looked at", busy: 1, + key: &builds.BuilderBuild{State: builds.BuildBuilding, By: "novox.builder", Step: "claimed"}}, {name: "build again: a key failed for a pin missing is refused, it waits for another fingerprint", again: true, refuse: "does not retry", key: &builds.BuilderBuild{State: builds.BuildFailed, Error: builds.NotPinned + "web"}}, {name: "build again: a key with no build is refused", again: true, refuse: "has no build key"}, @@ -1047,6 +1057,9 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { r.b.hands[r.key] = &handWatch{} } r.b.known["m"] = m // the builder has read the modules value, and looks at nothing until asked + if c.standing { + r.b.tried[r.key] = true // its look of the key ended before its write: a later look takes over nothing by itself + } log := &logged{} r.b.Logf = log.logf res := r.b.BuildAgain(context.Background(), "m", fp1, "no tun on ace", "shanks.lead", 2*time.Second) @@ -1075,7 +1088,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) { return } asks := r.asks.Asks() - if res.Refused != "" || res.Pending != "" || res.Revision <= rev || len(asks) != 1 || res.Asks[buildstep.AskGo] != asks[0].ID { + if res.Refused != "" || res.Pending != "" || res.Revision < rev || (res.Revision == rev && !c.standing) || len(asks) != 1 || res.Asks[buildstep.AskGo] != asks[0].ID { t.Fatalf("answer %+v after %d asks; want the new revision (was %d) and the ask id of the one ask", res, len(asks), rev) } if v := valueOf(t, r.st, r.key); v.State != builds.BuildDone || after <= res.Revision { @@ -1350,6 +1363,24 @@ func TestAnAskTheKeyNamesAndNobodyPutIsPutAtStart(t *testing.T) { t.Errorf("held %v: the key = %+v, want done", held, v) } } + // Issue 1966 (ADR 0668 §1, §5; 1748 note 27): the note of the ask and its put are one act. A note a deadline kept from the + // key fails the key in the transport fault's words: the builder does not take it again by itself (ADR 0664 §3), and + // builder_build_again builds it. + r := newRig(t, goBundle) + r.st.faults = []error{fmt.Errorf("the key: %w", context.DeadlineExceeded)} + if v := r.consider(t, "modules/m"); v.State != builds.BuildFailed || !builds.FailedOnTransportFault(v.Error) || len(r.asks.Asks()) != 0 { + t.Fatalf("a note the bus did not answer: the key = %+v after %d asks, want failed on a transport fault, nothing put", v, len(r.asks.Asks())) + } + if v := r.consider(t, "modules/m"); v.State != builds.BuildFailed || len(r.asks.Asks()) != 0 { + t.Fatalf("looked at again by itself: the key = %+v after %d asks, want it left failed", v, len(r.asks.Asks())) + } + if res := r.b.BuildAgain(context.Background(), "m", fp1, "the bus was back", "shanks.lead", 2*time.Second); res.Refused != "" || len(r.asks.Asks()) != 1 { + t.Fatalf("built again by hand: %+v after %d asks", res, len(r.asks.Asks())) + } + r.b.Wait() + if v := valueOf(t, r.st, r.key); v.State != builds.BuildDone { + t.Errorf("built again by hand: the key = %+v, want done", v) + } } // unreadableHeld cannot tell whether an ask is held.