Start no check while a delivery builds: the checker reads the controller's bucket of the plans in their build phase, absent as empty (issue 1685)
This commit is contained in:
@@ -0,0 +1,76 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"sync"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
// holdWatch holds the checker back while a delivery builds (ADR 0484 clause 12): it watches the controller's bucket of the
|
||||
// delivery plans in their build phase and reads it as empty where it is absent (issue 1652 makes it) or cannot be read,
|
||||
// saying so once. The bucket may come into being after the checker started and the checker holds no timer (ADR 0437 §8), so
|
||||
// each time a check is about to start and the bucket is not watched yet, the checker looks for it.
|
||||
type holdWatch struct {
|
||||
ctx context.Context
|
||||
js bucketOpener
|
||||
hold check.PlanHold
|
||||
changed func(ctx context.Context) // the last plan ended: the checks that waited may start
|
||||
logger *log.Logger
|
||||
errs chan error
|
||||
mu sync.Mutex
|
||||
watched bool
|
||||
said bool
|
||||
}
|
||||
|
||||
func newHoldWatch(ctx context.Context, js bucketOpener, changed func(context.Context), logger *log.Logger) *holdWatch {
|
||||
return &holdWatch{ctx: ctx, js: js, changed: changed, logger: logger, errs: make(chan error, 1)}
|
||||
}
|
||||
|
||||
// Held says whether a delivery builds now.
|
||||
func (w *holdWatch) Held() bool {
|
||||
w.ensure()
|
||||
return w.hold.Held()
|
||||
}
|
||||
|
||||
func (w *holdWatch) say(format string, args ...any) {
|
||||
if !w.said {
|
||||
w.said = true
|
||||
w.logger.Printf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// ensure starts the watch if the bucket exists and is not watched.
|
||||
func (w *holdWatch) ensure() {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if w.watched {
|
||||
return
|
||||
}
|
||||
kv, err := w.js.KeyValue(w.ctx, check.PlansBuildingBucket)
|
||||
switch {
|
||||
case errors.Is(err, jetstream.ErrBucketNotFound):
|
||||
w.say("the bucket %s does not exist yet: it is read as empty, so no check is held back", check.PlansBuildingBucket)
|
||||
return
|
||||
case err != nil:
|
||||
w.say("the bucket %s cannot be read, so it is read as empty: %v", check.PlansBuildingBucket, err)
|
||||
return
|
||||
}
|
||||
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.hold.Put(e.Key(), !opOf(e)) })
|
||||
if err != nil {
|
||||
w.say("the bucket %s cannot be watched, so it is read as empty: %v", check.PlansBuildingBucket, err)
|
||||
return
|
||||
}
|
||||
w.watched, w.said = true, false
|
||||
go func() {
|
||||
w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
|
||||
if w.hold.Put(e.Key(), !opOf(e)) && !w.hold.Held() {
|
||||
w.changed(w.ctx)
|
||||
}
|
||||
})
|
||||
}()
|
||||
}
|
||||
@@ -498,3 +498,31 @@ func TestACheckLeftRunningIsTakenOverWhenTheCheckerStarts(t *testing.T) { // N6/
|
||||
t.Errorf("verdict = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
// TestNoCheckStartsWhileADeliveryBuildsAndOneStartsWhenItEnds: the controller's bucket of the plans in their build phase
|
||||
// holds a plan's key; the candidate waits, no step is called; the key deleted, the check starts and passes (ADR 0484 clause 12).
|
||||
func TestNoCheckStartsWhileADeliveryBuildsAndOneStartsWhenItEnds(t *testing.T) { // N8/AC1
|
||||
b := startBed(t)
|
||||
plans, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_plans-building", History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := plans.Put(b.ctx, "plan-1", []byte(`{"plan":"plan-1","since":"2026-10-21T19:00:00Z","seen":1}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key := b.candidate(t, "feat/x", pull(repo, 7, head1))
|
||||
b.start(t, nil)
|
||||
time.Sleep(1500 * time.Millisecond) // a wait for nothing to happen: there is no event to wait on
|
||||
if _, err := b.buckets[delivery.CheckerVerdictsBucket].Get(b.ctx, key); err == nil {
|
||||
t.Fatal("a check started while a delivery builds")
|
||||
}
|
||||
if n := b.fake.count("mesh-build-go.build_go_fetch"); n != 0 {
|
||||
t.Fatalf("%d fetches while a delivery builds", n)
|
||||
}
|
||||
if err := plans.Delete(b.ctx, "plan-1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v := b.verdict(t, key, ended); v.State != delivery.VerdictPass {
|
||||
t.Errorf("verdict = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -198,6 +198,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Link: link,
|
||||
Logf: logger.Printf,
|
||||
}
|
||||
holds := newHoldWatch(ctx, js, ch.Reconsider, logger)
|
||||
ch.Hold = holds.Held
|
||||
announce := func(ctx context.Context, event, key string) {
|
||||
body, _ := json.Marshal(map[string]string{"key": key})
|
||||
if _, err := js.Publish(ctx, eventSubject(event), body); err != nil {
|
||||
@@ -333,5 +335,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
case err := <-watch.errs:
|
||||
return err
|
||||
case err := <-holds.errs:
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
package check
|
||||
|
||||
import "sync"
|
||||
|
||||
// PlansBuildingBucket is the controller's bucket of the delivery plans in their build phase, one key for each plan, written
|
||||
// by the controller alone (issue 1652): a plan key there means a delivery builds on the build nodes, and no new check starts
|
||||
// meanwhile (ADR 0484 clause 12, ADR 0483 clause 6). The bucket may be absent: the controller's issue 1652 is delivered
|
||||
// before R4c-5b, and if it is not, the bucket is read as empty.
|
||||
const PlansBuildingBucket = "mesh-controller_plans-building"
|
||||
|
||||
// PlanHold holds which delivery plans build now, from the controller's bucket (a watch puts the changes in). Safe for use
|
||||
// from the watch and from a check at once.
|
||||
type PlanHold struct {
|
||||
mu sync.Mutex
|
||||
plans map[string]bool
|
||||
}
|
||||
|
||||
// Put takes a change of the bucket in: a put of a plan's key, or its delete. It answers whether a delivery builds now
|
||||
// differently than before, so a watch wakes the checker when the last plan ended.
|
||||
func (h *PlanHold) Put(key string, deleted bool) (changed bool) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
before := len(h.plans) > 0
|
||||
if deleted {
|
||||
delete(h.plans, key)
|
||||
} else {
|
||||
if h.plans == nil {
|
||||
h.plans = map[string]bool{}
|
||||
}
|
||||
h.plans[key] = true
|
||||
}
|
||||
return before != (len(h.plans) > 0)
|
||||
}
|
||||
|
||||
// Held says whether any delivery plan builds now.
|
||||
func (h *PlanHold) Held() bool {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return len(h.plans) > 0
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func TestADeliveryThatBuildsHoldsTheChecksAndTheEndOfTheLastPlanSaysSo(t *testing.T) { // N8/AC1
|
||||
var h PlanHold
|
||||
if h.Held() {
|
||||
t.Fatal("an empty bucket holds")
|
||||
}
|
||||
if !h.Put("plan-1", false) || !h.Held() {
|
||||
t.Error("a first plan did not hold")
|
||||
}
|
||||
if h.Put("plan-2", false) {
|
||||
t.Error("a second plan changed whether a delivery builds")
|
||||
}
|
||||
if h.Put("plan-1", true) || !h.Held() {
|
||||
t.Error("one of two plans ending released the hold")
|
||||
}
|
||||
if !h.Put("plan-2", true) || h.Held() {
|
||||
t.Error("the last plan ending did not release the hold")
|
||||
}
|
||||
if h.Put("plan-3", true) {
|
||||
t.Error("the delete of a plan never put changed the hold")
|
||||
}
|
||||
}
|
||||
|
||||
func TestARunningCheckIsNotStoppedWhenADeliveryStartsToBuild(t *testing.T) { // ADR 0484 clause 12, ADR 0483 clause 4
|
||||
ctx := context.Background()
|
||||
var h PlanHold
|
||||
j := &judgeFunc{f: func(_ context.Context, _ *Case, call int) (Outcome, error) {
|
||||
if call == 1 {
|
||||
h.Put("plan-1", false) // a delivery starts while this check runs
|
||||
}
|
||||
return Outcome{Verdict: passing()}, nil
|
||||
}}
|
||||
r := newRig(t, j)
|
||||
r.ch.Hold = h.Held
|
||||
r.ch.Candidate(ctx, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1)))
|
||||
r.ch.Wait()
|
||||
if v := r.verdict(t, keyOf(t, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1)))); v.State != delivery.VerdictPass {
|
||||
t.Errorf("verdict = %+v, want the running check to end as it was", v)
|
||||
}
|
||||
// A new candidate waits while the plan builds.
|
||||
r.ch.Candidate(ctx, "feat/y", candidateOf(t, "feat/y", pullOf(repoA, 8, head2)))
|
||||
r.ch.Wait()
|
||||
if j.count() != 1 {
|
||||
t.Errorf("judged %d times while a delivery builds, want only the one already running", j.count())
|
||||
}
|
||||
h.Put("plan-1", true)
|
||||
r.ch.Reconsider(ctx)
|
||||
r.ch.Wait()
|
||||
if j.count() != 2 {
|
||||
t.Errorf("judged %d times after the delivery ended, want the waiting candidate too", j.count())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user