jschoubben
c32474ee40
mesh/delivery superseded: a newer head of the same pull request
mesh/check failed: integration: mesh-delivery-gate/cmd/mesh-d… | c32474ee40f4 on 3bec357968da | judged by the candidate's mesh-build-go at 2e667ac15ce7
113 lines
4.1 KiB
Go
113 lines
4.1 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
)
|
|
|
|
// checkCommit is the commit the check ask fetches, apart from the delivery ask's, so the fetch order names which ran first.
|
|
const checkCommit = "89abcdef0123456789abcdef0123456789abcdef"
|
|
|
|
// 1887/AC-order, 1887/AC-pull against the real bus: a check ask published before a delivery ask runs after it, and at
|
|
// capacity 1 the second waits in the stream until the first ends.
|
|
//
|
|
// **A bus of its own.** The integration level runs every directory's tests at once on one bus, and this test names the
|
|
// build seat's stream and its fixed workers (SEAT_NODE_BUILD_GO and its workers) and deletes the stream at its end, so
|
|
// anything else on that bus, a puller or a leftover consumer, could hold or take its delivery ask and let the check
|
|
// ask run first. It starts from testbus (mesh-sdk, novox/hq ADR 0495 clause 5): a bus for this test alone, gone after it.
|
|
//
|
|
// **The order is read from the record.** At capacity 1 the puller starts the check the instant the delivery ask's step
|
|
// ends, well before any poll, so "the check key holds a state" is true on correct behaviour (issue 1887). The test
|
|
// waits for both asks to end and compares the step records' own Started and Ended times, and the fetch order by commit.
|
|
func TestTheModuleTakesADeliveryAskBeforeAnEarlierCheck(t *testing.T) {
|
|
nc, err := nats.Connect(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, stop := context.WithCancel(context.Background())
|
|
defer stop()
|
|
if err := buildstep.DeclareAskWorkers(ctx, js, delivery.AskGo, 30*time.Second); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = js.DeleteStream(context.Background(), "SEAT_NODE_BUILD_GO") })
|
|
check := deliveryAsk()
|
|
check.Kind = delivery.AskCheck
|
|
check.Build.Fetch["commit"] = checkCommit
|
|
first := deliveryAsk()
|
|
for _, a := range []delivery.StepAsk{check, first} {
|
|
raw, _ := delivery.EncodeStepAsk(a)
|
|
subject, _ := delivery.AskSubject(a.Language, a.Kind)
|
|
if _, err := js.Publish(ctx, subject, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
states := newStates()
|
|
m := moduleWith(states, "go1.26.0")
|
|
m.Capacity = func() int { return 1 }
|
|
var mu sync.Mutex
|
|
var order []string
|
|
steps := fakeSteps(new([]string))
|
|
for i := range steps {
|
|
if steps[i].Name == "build_go_fetch" {
|
|
run := steps[i].Run
|
|
steps[i].Run = func(ctx context.Context, args map[string]any) (any, error) {
|
|
mu.Lock()
|
|
order = append(order, args["commit"].(string))
|
|
mu.Unlock()
|
|
return run(ctx, args)
|
|
}
|
|
}
|
|
}
|
|
p := m.Puller(steps, &buildstep.JetStreamAsks{JS: js, Language: delivery.AskGo})
|
|
p.PullWait = 200 * time.Millisecond
|
|
go p.Pull(ctx)
|
|
var ran map[string][]delivery.Step
|
|
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(10 * time.Millisecond) {
|
|
ran = map[string][]delivery.Step{}
|
|
ended := 0
|
|
states.mu.Lock()
|
|
for _, a := range []delivery.StepAsk{first, check} {
|
|
key, _ := delivery.AskStepKey("ace", a.ID)
|
|
v := append([]delivery.Step(nil), states.steps[key]...)
|
|
ran[a.ID] = v
|
|
if len(v) > 0 && v[len(v)-1].State != delivery.StepRunning {
|
|
ended++
|
|
}
|
|
}
|
|
states.mu.Unlock()
|
|
if ended == 2 {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("the asks never both ended: %v", ran)
|
|
}
|
|
}
|
|
// The check's step may start the moment the delivery's ends, so the order is read from the records' own times.
|
|
deliveryEnded := ran[first.ID][len(ran[first.ID])-1].Ended
|
|
checkStarted := ran[check.ID][0].Started
|
|
if checkStarted.Before(deliveryEnded) {
|
|
t.Fatalf("the check ask started at %v, before the delivery ask ended at %v at capacity 1", checkStarted, deliveryEnded)
|
|
}
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if len(order) != 2 || order[0] != askCommit || order[1] != checkCommit {
|
|
t.Fatalf("fetched %v, want the delivery ask's commit %s first, then the check's %s", order, askCommit, checkCommit)
|
|
}
|
|
}
|