jschoubben
d29e1426d0
A-163: the asker in `accept.<queue>.<asker>` is the asking seat's name (mesh-pipeline-runner, mesh-catalog, mesh-wants, mesh-builds), never a nickname: the builder asks as mesh-builds (was "builder"), the indexer's validation as mesh-catalog (was "indexer"), the assigner's runs as mesh-wants (was "assigner"), the runner's steps on `accept.check.mesh-pipeline-runner` and `accept.delivery.mesh-pipeline-runner` (were `accept.check.step` and `accept.delivery.step`, where the holder read `step` as the asker). An asker naming none (the checker) asks on main's subject (A-184). A build seat's queues are delivery, check and recheck alone (A-170, A-184): buildstep's AskKinds holds no step, a holder takes a step from check or delivery, and the grants part makes a worker per queue, never per kind. A holder keys a named ask's step `<asker>.<node>.<id>` and main's `<node>.<id>` (§481, §483); readers watch their own name's keys. The composition step's step-progress names the one caller mesh-wants' definition lists for step, read from the seat list (A-294). Builds: design 47 G §481, §483, §543, §600, A-163, A-170, A-184, A-248, A-294 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
63 lines
2.1 KiB
Go
63 lines
2.1 KiB
Go
//go:build integration
|
|
|
|
package buildrun
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
)
|
|
|
|
// 1887/A4a, 1887/AC3 on the real bus: an ask put twice under one id is one message, and the asker reads the step a node
|
|
// wrote under `<node>.<ask id>`.
|
|
func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
|
|
nc, err := nats.Connect(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
js, _ := jetstream.New(nc)
|
|
ctx := context.Background()
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_NODE_BUILD_GO", Subjects: []string{"mesh.seat.node-build-go.accept.>"},
|
|
Retention: jetstream.WorkQueuePolicy}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: StepsBucket("mesh-build-go")})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
bus := JetStreamAsks{JS: js}
|
|
// §481: a named asker reads its step under its own name, `<asker>.<node>.<id>`.
|
|
k := Asker{Name: "mesh-builds", Seats: testSeats(), Publisher: bus, Waiter: bus, Bound: 5 * time.Second}
|
|
a := goAsk("mvl0frv18d1960")
|
|
for i := 0; i < 2; i++ {
|
|
if err := k.Put(ctx, a); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
s, _ := js.Stream(ctx, "SEAT_NODE_BUILD_GO")
|
|
if info, _ := s.Info(ctx); info.State.Msgs != 1 {
|
|
t.Fatalf("%d messages for one ask put twice", info.State.Msgs)
|
|
}
|
|
go func() {
|
|
time.Sleep(100 * time.Millisecond)
|
|
key, _ := buildstep.AskStepKey("mesh-builds", "ace", a.ID)
|
|
running, _ := buildstep.EncodeStep(buildstep.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: buildstep.StepRunning, Started: time.Now()})
|
|
_, _ = kv.Put(ctx, key, running)
|
|
passed, _ := buildstep.EncodeStep(buildstep.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: buildstep.StepPassed, Started: time.Now(),
|
|
Result: json.RawMessage(`{"artifacts":[]}`)})
|
|
_, _ = kv.Put(ctx, key, passed)
|
|
}()
|
|
step, err := k.Passed(k.Wait(ctx, a))
|
|
if err != nil || step.Node != "ace" {
|
|
t.Fatalf("step %+v, %v", step, err)
|
|
}
|
|
}
|