novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Asker rename R2: every asker of a build seat asks under its seat's own name, and its step is keyed so

Browse Source 
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
This commit is contained in:
jschoubben
2026-11-12 18:37:29 +01:00
parent 4851475773
commit d29e1426d0
35 changed files with 286 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go
+3 -3
View File
@@ -30,9 +30,9 @@ import (
// (machine-away) is not composed and passes with a warning (§64).
const (
// runAsker is the assigner's name as an asker of a build seat: the last token of the subject its run asks are put on and
// the first of their step keys (§467, §481, §600).
runAsker = "assigner"
// runAsker is the assigner's name as an asker of a build seat, its seat's own name: the last token of the subject its run
// asks are put on and the first of their step keys (§467, §481, §600, A-163).
runAsker = meshwants.Seat
// changeCase is the one case each request composes.
changeCase = "change"
// inputVerbKey is the verb a run ask's input is served on: the seat mesh-wants' input (§354).
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep.go
+30 -6
View File
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"strings"
"sync"
"time"
@@ -10,6 +11,7 @@ import (
meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// The composition step (design 47 G §22, §354, §63): the assigner takes the seat mesh-wants' accepted kind `step`, the
@@ -32,15 +34,37 @@ type compositionSteps struct {
run func(ctx context.Context, id string, run meshwants.StepRun) stepOutcome
now func() time.Time
logf func(string, ...any)
// asker is the name the step's progress names as its asker.
asker string
// asker is the name the step's progress names as its asker, read when it is said (stepAskerOf).
asker func() string
// ticks is a ticker of every d, its channel and its stop (time.NewTicker); nil is the clock's. A test hands its own.
ticks func(d time.Duration) (<-chan time.Time, func())
}
// stepAsker is the asker a composition step's progress names: the pipeline runner, the one asker that sends the accepted kind
// step on mesh-wants (design 47 G A-267), whose accept subject names no asker.
const stepAsker = "runner"
// stepAskerOf is the asker a composition step's progress names: the one caller the seat mesh-wants' definition lists for its
// accepted kind step, read from the seat list when it is said, since that subject names no asker (design 47 G A-294, which
// corrects A-267). Empty while the seat list holds no definition naming exactly one.
func stepAskerOf(definition func() (seatdef.Definition, bool)) func() string {
return func() string {
d, ok := definition()
if !ok {
return ""
}
var one string
for _, m := range d.Accepts {
if m.Name != meshwants.AcceptStep || m.Callers == nil {
continue
}
for _, e := range m.Callers.List {
name, isSeat := strings.CutPrefix(e, seatdef.CallerSeatPrefix)
if !isSeat || (one != "" && one != name) {
return ""
}
one = name
}
}
return one
}
}
// progressEvery is how often a running composition step says its progress (design 47 G §496, A-262).
const progressEvery = time.Minute
@@ -115,7 +139,7 @@ func (c *compositionSteps) sayProgress(id string, phase *stepPhase) (stop func()
return
case <-tick:
at := c.now().UTC()
p := meshwants.StepProgressEvent{Ask: id, Asker: c.asker, Phase: phase.get(), At: at}
p := meshwants.StepProgressEvent{Ask: id, Asker: c.asker(), Phase: phase.get(), At: at}
if err := c.say(meshwants.EventStepProgress+"."+c.node, p, id+".progress."+at.Format(time.RFC3339Nano)); err != nil {
c.logf("saying the progress of the composition step %s: %v", id, err)
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go
+2 -1
View File
@@ -125,7 +125,8 @@ func TestTheCompositionStepComposesEachMachineThroughTheBuildSeat(t *testing.T)
// The stand-in Go build seat takes the run asks put under the assigner's name, reads each request by the verb input and says
// its end with the composer's verdict: n1 composes, n2 does not.
queue, _ := buildstep.AskStream(buildstep.AskGo)
subject, _ := buildstep.AskSubjectBy(buildstep.AskGo, buildstep.AskCheck, runAsker)
// A-163: the assigner asks under its seat's own name.
subject := "mesh.seat.node-build-go.accept.check.mesh-wants"
worker, err := b.js.CreateOrUpdateConsumer(b.ctx, queue, jetstream.ConsumerConfig{Durable: "stand-in", FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy})
if err != nil {
t.Fatal(err)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_test.go
+19 -2
View File
@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"sync"
"testing"
"time"
@@ -79,7 +80,23 @@ func TestARunningCompositionStepSaysItsProgress(t *testing.T) {
stopped := make(chan struct{})
at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
release := make(chan struct{})
c := &compositionSteps{node: "n1", say: said.say, now: func() time.Time { return at }, logf: t.Logf, asker: "runner",
// A-294: the asker is the one caller mesh-wants' definition lists for step, read from the seat list; none while unread
// or while it lists more than one.
runner := seatdef.Definition{Seat: meshwants.Seat, Accepts: []seatdef.Message{{Name: meshwants.AcceptStep,
Callers: &seatdef.Callers{List: []string{"seat:mesh-pipeline-runner"}}}}}
two := seatdef.Definition{Seat: meshwants.Seat, Accepts: []seatdef.Message{{Name: meshwants.AcceptStep,
Callers: &seatdef.Callers{List: []string{"seat:mesh-pipeline-runner", "seat:mesh-catalog"}}}}}
for _, d := range []struct {
def seatdef.Definition
read bool
want string
}{{runner, true, "mesh-pipeline-runner"}, {runner, false, ""}, {two, true, ""}} {
if got := stepAskerOf(func() (seatdef.Definition, bool) { return d.def, d.read })(); got != d.want {
t.Errorf("the asker of %+v (read %v) = %q, want %q", d.def.Accepts, d.read, got, d.want)
}
}
c := &compositionSteps{node: "n1", say: said.say, now: func() time.Time { return at }, logf: t.Logf,
asker: stepAskerOf(func() (seatdef.Definition, bool) { return runner, true }),
ticks: func(d time.Duration) (<-chan time.Time, func()) {
if d != time.Minute {
t.Errorf("the progress ticks every %s, want a minute", d)
@@ -105,7 +122,7 @@ func TestARunningCompositionStepSaysItsProgress(t *testing.T) {
said.mu.Lock()
p, ok := said.said[progress].(meshwants.StepProgressEvent)
said.mu.Unlock()
if !ok || p.Ask != "ask-1" || p.Asker != "runner" || p.Phase != meshwants.StepProgressEventPhaseRun || !p.At.Equal(at) {
if !ok || p.Ask != "ask-1" || p.Asker != "mesh-pipeline-runner" || p.Phase != meshwants.StepProgressEventPhaseRun || !p.At.Equal(at) {
t.Fatalf("step-progress %+v", p)
}
close(release)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+2 -1
View File
@@ -665,7 +665,8 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
if bus != nil && js != nil && inputs != nil && p.publishWants != nil {
run := &compositionRun{inputs: &compositionInputs{machines: w.machinesOf, call: bus.CallKeptOn}, keeper: inputs, call: bus.CallKept,
trunk: trunkOf(iv, w.trunkName), running: w.runningOf, away: away, put: runAskPublisher(js), ends: ends, ask: bus.CallFile, now: time.Now, logf: logger.Printf}
steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf, asker: stepAsker}
steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf,
asker: stepAskerOf(seatlist.Definition(gate.Reader, meshwants.Seat))}
if err := serveCompositionSteps(ctx, nc, steps); err != nil {
logger.Printf("the composition step is not heard, so no composition step runs until the assigner starts again: %v", err)
} else {
modules/mesh-delivery-builder/buildrun/asks.go
+10 -11
View File
@@ -1,7 +1,6 @@
package buildrun
import (
"cmp"
"context"
"encoding/json"
"errors"
@@ -30,13 +29,13 @@ type AskPublisher interface {
// AskWaiter watches an ask's steps: every value a key `*.<ask id>` of the build module's steps state takes, running or
// ended, on the channel it answers, until ctx ends. A watch, not a poll.
type AskWaiter interface {
WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error)
WatchAsk(ctx context.Context, module, asker, askID string) (<-chan buildstep.Step, error)
}
// Asker puts asks and waits for their steps.
type Asker struct {
// Name is the asker the subject names, the caller's seat name (design 47 G A-248, A-163): empty puts under the builder's
// name until the asker rename after the join.
// Name is the asker the subject names, the caller's seat name (design 47 G A-248, A-163): empty puts on main's subject,
// which names none, as the checker asks until main's subject goes (A-170, A-184); the steps are read under the same name.
Name string
Publisher AskPublisher
Waiter AskWaiter
@@ -181,8 +180,8 @@ func (k Asker) dropped(a buildstep.StepAsk, why string) {
// 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, under the asker's Name, else the builder's
// name (design 47 G §481, §600, A-248: `accept.<queue>.builder`). The put is a kept call (novox/hq ADR 0668
// Put puts a on its stream: the seat's accept subject of its language and queue, under the asker's Name, else on main's
// subject (design 47 G §481, §600, A-163, A-248: `accept.<queue>.<seat>`). 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 `transport fault:` (TransportFault), so the key it
@@ -192,7 +191,7 @@ func (k Asker) Put(ctx context.Context, a buildstep.StepAsk) error {
if err != nil {
return err
}
subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, cmp.Or(k.Name, buildstep.AskerBuilder))
subject, err := buildstep.AskSubjectBy(a.Language, a.Kind, k.Name)
if err != nil {
return err
}
@@ -229,7 +228,7 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
module := bm.Module
watch, cancel := context.WithCancel(ctx)
defer cancel()
steps, err := k.Waiter.WatchAsk(watch, module, a.ID)
steps, err := k.Waiter.WatchAsk(watch, module, k.Name, a.ID)
if err != nil {
return buildstep.Step{}, err
}
@@ -487,8 +486,8 @@ func (s JetStreamAsks) openKept(ctx context.Context, module, askID, pattern stri
// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
// once, and every change after. A watch that cannot be opened for a transport fault is opened again once the bus answers, and one
// that closes is taken up again from the revision after the last it handed over (design 47 G §568), so a fault fails no key.
func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
pattern, err := buildstep.AskStepWatch(askID)
func (s JetStreamAsks) WatchAsk(ctx context.Context, module, asker, askID string) (<-chan buildstep.Step, error) {
pattern, err := buildstep.AskStepWatch(asker, askID)
if err != nil {
return nil, err
}
@@ -582,7 +581,7 @@ func (f *FakeAsks) Write(step buildstep.Step) {
func (f *FakeAsks) End(step buildstep.Step) { f.Write(step) }
// WatchAsk is AskWaiter's: what was written so far, then every write.
func (f *FakeAsks) WatchAsk(ctx context.Context, _ string, askID string) (<-chan buildstep.Step, error) {
func (f *FakeAsks) WatchAsk(ctx context.Context, _, _ string, askID string) (<-chan buildstep.Step, error) {
out := make(chan buildstep.Step, 64)
f.mu.Lock()
if f.steps == nil {
modules/mesh-delivery-builder/buildrun/asks_integration_test.go
+3 -2
View File
@@ -34,7 +34,8 @@ func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
t.Fatal(err)
}
bus := JetStreamAsks{JS: js}
k := Asker{Seats: testSeats(), Publisher: bus, Waiter: bus, Bound: 5 * time.Second}
// §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 {
@@ -47,7 +48,7 @@ func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
}
go func() {
time.Sleep(100 * time.Millisecond)
key, _ := buildstep.AskStepKey("ace", a.ID)
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(),
modules/mesh-delivery-builder/buildrun/asks_test.go
+8 -8
View File
@@ -40,13 +40,13 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
if err != nil || step.Node != "ace" || len(f.Asks()) != 1 || len(seen) != 1 || seen[0].Node != "ace" || len(also) != 1 {
t.Fatalf("step %+v, %v, asks %d, seen %+v and %+v", step, err, len(f.Asks()), seen, also)
}
// Design 47 G §481, §600: the ask is put under the builder's name, accept.<queue>.builder.
if len(subjects.seen) != 1 || subjects.seen[0] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind+".builder" {
// Design 47 G §600, A-170, A-184: an asker that names none puts on main's subject, accept.<queue>, as the checker does.
if len(subjects.seen) != 1 || subjects.seen[0] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind {
t.Fatalf("the ask was put on %v", subjects.seen)
}
// A-248: an asker that names itself puts under its own name.
byName := Asker{Name: "indexer", Seats: testSeats(), Publisher: subjects, Waiter: f}
if err := byName.Put(context.Background(), goAsk("mvl0frv18d1959")); err != nil || subjects.seen[len(subjects.seen)-1] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind+".indexer" {
// A-163, A-248: an asker that names itself puts under its seat's own name.
byName := Asker{Name: "mesh-catalog", Seats: testSeats(), Publisher: subjects, Waiter: f}
if err := byName.Put(context.Background(), goAsk("mvl0frv18d1959")); err != nil || subjects.seen[len(subjects.seen)-1] != "mesh.seat.node-build-go.accept."+goAsk("x").Kind+".mesh-catalog" {
t.Fatalf("a named asker put on %v (%v)", subjects.seen, err)
}
f.Answer = func(a buildstep.StepAsk) buildstep.Step {
@@ -347,7 +347,7 @@ func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unl
// deafWatch is a watch of an ask's step that stays open and hears nothing.
type deafWatch struct{}
func (deafWatch) WatchAsk(context.Context, string, string) (<-chan buildstep.Step, error) {
func (deafWatch) WatchAsk(context.Context, string, string, string) (<-chan buildstep.Step, error) {
return make(chan buildstep.Step), nil
}
@@ -357,14 +357,14 @@ type closingWatch struct {
closed bool
}
func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
func (c *closingWatch) WatchAsk(ctx context.Context, module, asker, askID string) (<-chan buildstep.Step, error) {
if !c.closed {
c.closed = true
ch := make(chan buildstep.Step)
close(ch)
return ch, nil
}
return c.inner.WatchAsk(ctx, module, askID)
return c.inner.WatchAsk(ctx, module, asker, askID)
}
// seatDrops keeps the asks an Asker dropped: "<id> <why>".
modules/mesh-delivery-builder/buildrun/heardends.go
+8 -8
View File
@@ -22,7 +22,7 @@ import (
// StepReader reads, now, every step a build module's steps state holds for an ask: its keys `*.<ask id>`.
type StepReader interface {
AskSteps(ctx context.Context, module, askID string) ([]buildstep.Step, error)
AskSteps(ctx context.Context, module, asker, askID string) ([]buildstep.Step, error)
}
// reopenTries is how often a wait opens its ask's watch again after it closed, before the wait ends failed.
@@ -53,8 +53,8 @@ func (h *HeardEnds) logf(format string, args ...any) {
}
// WatchAsk is AskWaiter's: what Inner's watch hands, and, at each nudge, what the ask's keys hold now.
func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
in, err := h.Inner.WatchAsk(ctx, module, askID)
func (h *HeardEnds) WatchAsk(ctx context.Context, module, asker, askID string) (<-chan buildstep.Step, error) {
in, err := h.Inner.WatchAsk(ctx, module, asker, askID)
if err != nil {
return nil, err
}
@@ -95,7 +95,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
return
}
reopened++
if in, err = h.Inner.WatchAsk(ctx, module, askID); err != nil {
if in, err = h.Inner.WatchAsk(ctx, module, asker, askID); err != nil {
h.logf("the watch of the ask %s's step on %s closed and cannot be opened again, so its wait ends: %v", askID, module, err)
return
}
@@ -109,7 +109,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
if h.Read == nil {
continue
}
steps, err := h.Read.AskSteps(ctx, module, askID)
steps, err := h.Read.AskSteps(ctx, module, asker, askID)
if err != nil {
h.logf("the steps of the ask %s on %s cannot be read: %v", askID, module, err)
continue
@@ -150,8 +150,8 @@ func (h *HeardEnds) Waiting() int {
}
// AskSteps is StepReader's: the keys `*.<ask id>` of the build module's steps state, read now.
func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]buildstep.Step, error) {
pattern, err := buildstep.AskStepWatch(askID)
func (s JetStreamAsks) AskSteps(ctx context.Context, module, asker, askID string) ([]buildstep.Step, error) {
pattern, err := buildstep.AskStepWatch(asker, askID)
if err != nil {
return nil, err
}
@@ -244,7 +244,7 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st
}
// AskSteps is StepReader's: the last step written for the ask, if any.
func (f *FakeAsks) AskSteps(_ context.Context, _ string, askID string) ([]buildstep.Step, error) {
func (f *FakeAsks) AskSteps(_ context.Context, _, _ string, askID string) ([]buildstep.Step, error) {
f.mu.Lock()
defer f.mu.Unlock()
if s := f.steps[askID]; len(s) > 0 {
modules/mesh-delivery-builder/buildrun/lostask.go
+2 -2
View File
@@ -193,11 +193,11 @@ func (w LostAskWaiter) now() time.Time {
}
// WatchAsk is AskWaiter's.
func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
func (w LostAskWaiter) WatchAsk(ctx context.Context, module, asker, askID string) (<-chan buildstep.Step, error) {
if w.Hearing != nil && w.Hearing.Deaf() != "" {
w.Hearing.Again()
}
inner, err := w.Inner.WatchAsk(ctx, module, askID)
inner, err := w.Inner.WatchAsk(ctx, module, asker, askID)
if err != nil {
return nil, err
}
modules/mesh-delivery-builder/buildrun/steps_resume_test.go
+1 -1
View File
@@ -63,7 +63,7 @@ func TestTheWatchOfAStepResumesAfterAFault(t *testing.T) {
}}
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
steps, err := asks.WatchAsk(ctx, "mesh-build-go", "mvl0frv18d1958")
steps, err := asks.WatchAsk(ctx, "mesh-build-go", "", "mvl0frv18d1958")
if err != nil {
t.Fatalf("a transport fault opening the watch failed the wait: %v", err)
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go
+4 -3
View File
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds"
"sort"
"strings"
"time"
@@ -63,7 +64,7 @@ func (h heldOnTheBus) seen(ctx context.Context, a buildstep.StepAsk, later bool)
if err != nil {
return seen, fmt.Errorf("the bucket %s: %w", buildrun.StepsBucket(module), err)
}
pattern, err := buildstep.AskStepWatch(a.ID)
pattern, err := buildstep.AskStepWatch(meshbuilds.Seat, a.ID)
if err != nil {
return seen, err
}
@@ -119,7 +120,7 @@ func (h heldOnTheBus) seen(ctx context.Context, a buildstep.StepAsk, later bool)
// stream is the build seat's stream of language, its name and the subject of kind on it.
func (h heldOnTheBus) stream(ctx context.Context, language, kind string) (jetstream.Stream, string, string, error) {
subject, err := buildstep.AskSubjectBy(language, kind, buildstep.AskerBuilder)
subject, err := buildstep.AskSubjectBy(language, kind, meshbuilds.Seat)
if err != nil {
return nil, "", "", err
}
@@ -291,7 +292,7 @@ func (h heldOnTheBus) Queue(ctx context.Context, language string, limit int) (bu
return q, err
}
for _, kind := range buildstep.AskKinds {
subject, _ := buildstep.AskSubjectBy(language, kind, buildstep.AskerBuilder)
subject, _ := buildstep.AskSubjectBy(language, kind, meshbuilds.Seat)
var ahead []build.Queued
if err := onSubject(ctx, stream, subject, func(m build.Queued) bool {
ahead = append(ahead, m)
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go
+1 -1
View File
@@ -1050,7 +1050,7 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
held := heldOnTheBus{js: b.js, seats: testSeats()}
goAsk := sdkstep.StepAsk{ID: v.Asks[sdkstep.AskGo], Language: sdkstep.AskGo, Kind: sdkstep.AskDelivery}
stale := sdkstep.StepAsk{ID: "mvo53jcm70b5f7", Language: sdkstep.AskGo, Kind: sdkstep.AskDelivery}
subject, _ := sdkstep.AskSubjectBy(sdkstep.AskGo, sdkstep.AskDelivery, sdkstep.AskerBuilder)
subject, _ := sdkstep.AskSubjectBy(sdkstep.AskGo, sdkstep.AskDelivery, meshbuilds.Seat)
if _, err := b.js.Publish(b.ctx, subject, []byte("{}"), jetstream.WithMsgID(stale.ID)); err != nil {
t.Fatal(err)
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+1 -1
View File
@@ -182,7 +182,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
b = &build.Builder{
Claimer: build.Claimer{Store: kvStore{buildsKV}, By: cred.Node + "." + cred.Module},
Seats: seats,
Asks: buildrun.Asker{Seats: seats, Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound, Dropper: buildrun.SeatDrops{Caller: calls},
Asks: buildrun.Asker{Name: meshbuilds.Seat, Seats: seats, Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound, Dropper: buildrun.SeatDrops{Caller: calls},
Bounds: &stepBounds{bounds: pubBoundsKV.KV, js: js, now: time.Now, logf: logger.Printf}, Expiry: stepExpiry},
Held: heldOnTheBus{js: js, seats: seats},
// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
modules/mesh-delivery-builder/internal/build/run_test.go
+1 -1
View File
@@ -1373,7 +1373,7 @@ func TestATakeOverThatCannotReadTheStreamWaitsForTheAsk(t *testing.T) {
// silentWatch is a watch of an ask's step that stays open and hears nothing.
type silentWatch struct{}
func (silentWatch) WatchAsk(context.Context, string, string) (<-chan buildstep.Step, error) {
func (silentWatch) WatchAsk(context.Context, string, string, string) (<-chan buildstep.Step, error) {
return make(chan buildstep.Step), nil
}
modules/mesh-delivery-checker/internal/check/lostask_test.go
+3 -3
View File
@@ -192,17 +192,17 @@ func (tr *troubled) PublishAsk(ctx context.Context, subject, id string, data []b
return nil
}
func (tr *troubled) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
func (tr *troubled) WatchAsk(ctx context.Context, module, asker, askID string) (<-chan buildstep.Step, error) {
tr.mu.Lock()
deaf := tr.deafIDs[askID]
tr.mu.Unlock()
if deaf {
return make(chan buildstep.Step), nil // open, and silent for good
}
return tr.FakeAsks.WatchAsk(ctx, module, askID)
return tr.FakeAsks.WatchAsk(ctx, module, asker, askID)
}
func (tr *troubled) AskSteps(_ context.Context, _, askID string) ([]buildstep.Step, error) {
func (tr *troubled) AskSteps(_ context.Context, _, _, askID string) ([]buildstep.Step, error) {
tr.mu.Lock()
defer tr.mu.Unlock()
if s, ok := tr.steps[askID]; ok {
modules/mesh-delivery-checker/internal/check/try_test.go
+1 -1
View File
@@ -267,7 +267,7 @@ func TestATryIsJudgedAsAGroupsCheckAndOwesTheForgeNothing(t *testing.T) { // ADR
t.Errorf("the ended try's steps = %+v, asks %v", got.Steps, got.Asks)
}
for _, a := range got.Asks { // the delivery view counts their steps as a try's (build_usage)
if _, err := buildstep.AskStepKey("shanks", a); err != nil {
if _, err := buildstep.AskStepKey("", "shanks", a); err != nil {
t.Errorf("the try names %q, which is no ask id: %v", a, err)
}
}
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go
+2 -2
View File
@@ -485,8 +485,8 @@ func (s *standIn) asked() []sdkstep.StepAsk {
func standInBuildSeat(t *testing.T, ctx context.Context, js jetstream.JetStream, refusing *sync.Map) *standIn {
t.Helper()
subject, err := sdkstep.AskSubjectBy(sdkstep.AskGo, sdkstep.AskCheck, index.ValidationAsker)
if err != nil {
t.Fatal(err)
if err != nil || subject != "mesh.seat.node-build-go.accept.check.mesh-catalog" {
t.Fatalf("the validation's subject %q %v, want the indexer's seat name its asker (A-163, A-232)", subject, err)
}
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_NODE_BUILD_GO_CHECK", Subjects: []string{"mesh.seat.node-build-go.accept.check.>"}}); err != nil {
t.Fatal(err)
modules/mesh-delivery-indexer/internal/index/validate.go
+3 -3
View File
@@ -25,9 +25,9 @@ const (
// validationProgram is the manifest validator's program, by its directory under the indexer entry's Go module: the indexer's
// own code, so no repository or module name is fixed here (ADR 0666 §5).
validationProgram = "cmd/mesh-judge"
// ValidationAsker is the asker the ask is put under (buildrun.Asker.Name, A-248). A-232 names the indexer's seat name mesh-catalog; the ask code still
// names askers by their nicknames until the asker rename after the join (A-163), so this is `indexer` until then.
ValidationAsker = "indexer"
// ValidationAsker is the asker the ask is put under (buildrun.Asker.Name, A-248): the indexer's seat name, mesh-catalog
// (design 47 G A-163, A-232).
ValidationAsker = meshcatalog.Seat
// validationCatalogSeat is the seat the indexer entry claims.
validationCatalogSeat = "mesh-catalog"
// The cases of the request, as the manifest validator names them.
modules/mesh-delivery-pipeline/internal/pipeline/place.go
+12 -9
View File
@@ -23,21 +23,24 @@ type StepAsker interface {
AskStep(accept, id string, body any) error
}
// acceptStep is where each build seat is sent its step, from its contract.
// acceptStep is where each seat is sent its step, from its contract: a build seat's on its queue `check` under the runner's
// own seat name, `accept.check.mesh-pipeline-runner` (design 47 G §600, A-163, A-170); mesh-builds and mesh-wants, which are
// no build seats, on their accept subject, which names no asker (A-294).
var acceptStep = map[string]string{
builds.Seat: builds.AcceptSubjectStep,
wants.Seat: wants.AcceptSubjectStep,
nodebuildgo.Seat: nodebuildgo.AcceptSubjectCheckStep,
nodebuildnode.Seat: nodebuildnode.AcceptSubjectCheckStep,
nodebuildimage.Seat: nodebuildimage.AcceptSubjectCheckStep,
nodebuildgo.Seat: nodebuildgo.AcceptSubjectCheckStep + "." + seat.Seat,
nodebuildnode.Seat: nodebuildnode.AcceptSubjectCheckStep + "." + seat.Seat,
nodebuildimage.Seat: nodebuildimage.AcceptSubjectCheckStep + "." + seat.Seat,
}
// acceptRecovery is where a recovery's step is sent on the build seats' stream `delivery`, ahead of the first round (§86, §415);
// the seats mesh-builds and mesh-wants have no such stream, so their steps stay on their own accept subject.
// acceptRecovery is where a recovery's step is sent on the build seats' queue `delivery`, under the runner's seat name, ahead
// of the first round (§86, §415, A-170); the seats mesh-builds and mesh-wants have no such queue, so their steps stay on their
// own accept subject.
var acceptRecovery = map[string]string{
nodebuildgo.Seat: nodebuildgo.AcceptSubjectDeliveryStep,
nodebuildnode.Seat: nodebuildnode.AcceptSubjectDeliveryStep,
nodebuildimage.Seat: nodebuildimage.AcceptSubjectDeliveryStep,
nodebuildgo.Seat: nodebuildgo.AcceptSubjectDeliveryStep + "." + seat.Seat,
nodebuildnode.Seat: nodebuildnode.AcceptSubjectDeliveryStep + "." + seat.Seat,
nodebuildimage.Seat: nodebuildimage.AcceptSubjectDeliveryStep + "." + seat.Seat,
}
// buildSteps are the steps an artifact's build seat runs, in the order they are asked (design 47 G §22).
modules/mesh-delivery-pipeline/internal/pipeline/place_test.go
+1 -1
View File
@@ -89,7 +89,7 @@ func TestStartPlacesTheStepsThePolicyNamesAndAsksEachBuildSeat(t *testing.T) {
t.Errorf("the build ask = %+v at %s", asks.asks[0], asks.at[0])
}
a := asks.asks[1].(nodebuildgo.StepAsk)
if asks.at[1] != nodebuildgo.AcceptSubjectCheckStep || a.Kind != "step" || a.Asker != "the-runner" || a.ID != p.Steps[1].Ask || asks.ids[1] != a.ID ||
if asks.at[1] != "mesh.seat.node-build-go.accept.check.mesh-pipeline-runner" || a.Kind != "step" || a.Asker != "the-runner" || a.ID != p.Steps[1].Ask || asks.ids[1] != a.ID ||
a.Step.Commit != "c1" || a.Step.Fingerprint != "src2:aa" || a.Step.Pipeline != "main.4.app" || a.Step.Artifact != "app:program" ||
a.Step.Step != "lint" || a.Step.Fetch.Repository != "o/app" || a.Step.Fetch.Dir != "modules/app" || a.Step.Revision != 4 {
t.Errorf("the ask = %+v at %s", a, asks.at[1])
modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go
+1 -1
View File
@@ -356,7 +356,7 @@ func TestARecoverysStepsAreAskedOnTheStreamDelivery(t *testing.T) {
for _, c := range []struct {
recovery bool
want string
}{{true, nodebuildgo.AcceptSubjectDeliveryStep + "," + wants.AcceptSubjectStep}, {false, nodebuildgo.AcceptSubjectCheckStep + "," + wants.AcceptSubjectStep}} {
}{{true, "mesh.seat.node-build-go.accept.delivery.mesh-pipeline-runner," + wants.AcceptSubjectStep}, {false, "mesh.seat.node-build-go.accept.check.mesh-pipeline-runner," + wants.AcceptSubjectStep}} {
var to []string
asker := askerFunc(func(accept, _ string, _ any) error { to = append(to, accept); return nil })
if err := askAll(asker, nil, "k", asks, c.recovery); err != nil || strings.Join(to, ",") != c.want {
modules/mesh-delivery-sdk/buildstep/ask.go
+49 -22
View File
@@ -33,8 +33,13 @@ const (
AskStep = "step"
)
// AskKinds are the accepted kinds of a build seat, in the order a holder pulls them (ADR 0536 §4).
var AskKinds = []string{AskDelivery, AskCheck, AskStep, AskRecheck}
// AskKinds are the queues of a build seat, in the order a holder pulls them (ADR 0536 §4): delivery, check and recheck,
// each one subject and one worker; a step rides check or delivery and a run check, the kind in the body alone (design 47 G
// §600, A-170, A-184).
var AskKinds = []string{AskDelivery, AskCheck, AskRecheck}
// askBodyKinds are the kinds an ask's body names: a queue's, or step.
var askBodyKinds = []string{AskDelivery, AskCheck, AskStep, AskRecheck}
// The languages of the build seats: each is one build module's (mesh-build-<language>) and one seat's
// (node-build-<language>).
@@ -108,12 +113,19 @@ const (
)
func checkAskKind(kind string) error {
if !slices.Contains(AskKinds, kind) {
if !slices.Contains(askBodyKinds, kind) {
return fmt.Errorf("buildstep: the ask kind %q is not delivery, check, step or recheck", kind)
}
return nil
}
func checkAskQueue(queue string) error {
if !slices.Contains(AskKinds, queue) {
return fmt.Errorf("buildstep: the queue %q is not delivery, check or recheck", queue)
}
return nil
}
func checkAskLanguage(language string) error {
if !slices.Contains(AskLanguages, language) {
return fmt.Errorf("buildstep: the ask language %q is not go, node or image", language)
@@ -129,27 +141,23 @@ func BuildSeat(language string) (string, error) {
return "node-build-" + language, nil
}
// AskSubject is the subject a build seat accepts a kind of ask on: the seat's accept subject of main's form, which names no
// AskSubject is the subject a build seat accepts a queue's asks on: the seat's accept subject of main's form, which names no
// asker (design 47 G §600, §483).
func AskSubject(language, kind string) (string, error) {
func AskSubject(language, queue string) (string, error) {
seat, err := BuildSeat(language)
if err != nil {
return "", err
}
if err := checkAskKind(kind); err != nil {
if err := checkAskQueue(queue); err != nil {
return "", err
}
return "mesh.seat." + seat + ".accept." + kind, nil
return "mesh.seat." + seat + ".accept." + queue, nil
}
// AskerBuilder is the builder's name as an asker: the last token of the subject its asks are put on and the first of its
// step keys (design 47 G §600, §481).
const AskerBuilder = "builder"
// AskSubjectBy is the subject a build seat accepts a queue's ask of asker on: `mesh.seat.<seat>.accept.<queue>.<asker>`, the
// asker's name the last token (design 47 G §481, §600). An empty asker is main's subject (AskSubject).
func AskSubjectBy(language, kind, asker string) (string, error) {
subject, err := AskSubject(language, kind)
// asker the asking seat's own name (design 47 G §481, §600, A-163). An empty asker is main's subject (AskSubject).
func AskSubjectBy(language, queue, asker string) (string, error) {
subject, err := AskSubject(language, queue)
if err != nil || asker == "" {
return subject, err
}
@@ -191,13 +199,13 @@ func AskStream(language string) (string, error) {
return "SEAT_" + strings.ToUpper(strings.ReplaceAll(seat, "-", "_")), nil
}
// AskWorker is the shared pull worker of a kind on a language's stream.
// AskWorker is the shared pull worker of a queue on a language's stream.
func AskWorker(language, kind string) (string, error) {
stream, err := AskStream(language)
if err != nil {
return "", err
}
if err := checkAskKind(kind); err != nil {
if err := checkAskQueue(kind); err != nil {
return "", err
}
return stream + "_" + strings.ToUpper(kind) + "_worker", nil
@@ -213,20 +221,39 @@ func AskNamedWorker(language, kind string) (string, error) {
return strings.TrimSuffix(worker, "_worker") + "_named_worker", nil
}
// AskStepKey is the key of the step of the ask askID on node.
func AskStepKey(node, askID string) (string, error) {
// AskStepKey is the key of the step of asker's ask askID on node: `<asker>.<node>.<id>`, the asker the one the ask's subject
// names, or main's `<node>.<id>` for an ask on main's subject, which names none (an empty asker; design 47 G §481, §483).
func AskStepKey(asker, node, askID string) (string, error) {
if err := CheckAskID(askID); err != nil {
return "", err
}
return StepKey(node, askID)
return StepKeyOf(asker, node, askID)
}
// AskStepWatch is the key pattern of an ask's step on any node.
func AskStepWatch(askID string) (string, error) {
// StepKeyOf is StepKey under asker: `<asker>.<node>.<id>`, or `<node>.<id>` for an empty asker (design 47 G §481).
func StepKeyOf(asker, node, id string) (string, error) {
key, err := StepKey(node, id)
if err != nil || asker == "" {
return key, err
}
if !askerOK.MatchString(asker) {
return "", fmt.Errorf("buildstep: %q is not an asker's name", asker)
}
return asker + "." + key, nil
}
// AskStepWatch is the key pattern of asker's ask's step on any node: `<asker>.*.<id>`, or `*.<id>` for main's form.
func AskStepWatch(asker, askID string) (string, error) {
if err := CheckAskID(askID); err != nil {
return "", err
}
return "*." + askID, nil
if asker == "" {
return "*." + askID, nil
}
if !askerOK.MatchString(asker) {
return "", fmt.Errorf("buildstep: %q is not an asker's name", asker)
}
return asker + ".*." + askID, nil
}
// StepAsk is one ask of a build module: the part's `StepAsk`.
modules/mesh-delivery-sdk/buildstep/ask_test.go
+23 -7
View File
@@ -268,11 +268,26 @@ func TestAskSubjectsStreamsAndKeys(t *testing.T) {
if _, err := AskSubject("rust", AskCheck); err == nil {
t.Error("an unknown language has a subject")
}
if k, err := AskStepKey("ace", testAskID); err != nil || k != "ace."+testAskID {
t.Errorf("%q %v", k, err)
// §481, §483: a named ask's step is keyed under its asker, main's under its node alone.
for _, c := range []struct{ asker, key, watch string }{
{"", "ace." + testAskID, "*." + testAskID},
{"mesh-pipeline-runner", "mesh-pipeline-runner.ace." + testAskID, "mesh-pipeline-runner.*." + testAskID},
} {
if k, err := AskStepKey(c.asker, "ace", testAskID); err != nil || k != c.key {
t.Errorf("%q: key %q %v", c.asker, k, err)
}
if w, _ := AskStepWatch(c.asker, testAskID); w != c.watch {
t.Errorf("%q: watch %q", c.asker, w)
}
if a, n, id, err := ParseStepKey(c.key); err != nil || a != c.asker || n != "ace" || id != testAskID {
t.Errorf("%q: parsed %q %q %q %v", c.key, a, n, id, err)
}
}
if w, _ := AskStepWatch(testAskID); w != "*."+testAskID {
t.Errorf("watch %q", w)
if _, err := AskStepKey("Bad.Name", "ace", testAskID); err == nil {
t.Error("a bad asker has a step key")
}
if _, err := AskSubject(AskGo, AskStep); err == nil {
t.Error("step is a queue: it rides check or delivery, its kind in the body (A-170, A-184)")
}
if k, _ := PauseKey("ace"); k != "paused.ace" {
t.Errorf("pause key %q", k)
@@ -606,12 +621,13 @@ func TestFetchBuildsResolvedAndToolchainDigestAreRead(t *testing.T) {
// §481, §600: a queue's ask of an asker is put on accept.<queue>.<asker>, and both forms parse.
func TestAskSubjectBy(t *testing.T) {
got, err := AskSubjectBy(AskGo, AskDelivery, AskerBuilder)
if err != nil || got != "mesh.seat.node-build-go.accept.delivery.builder" {
got, err := AskSubjectBy(AskGo, AskDelivery, "mesh-builds")
if err != nil || got != "mesh.seat.node-build-go.accept.delivery.mesh-builds" {
t.Fatalf("%q %v", got, err)
}
for _, c := range []struct{ subject, language, kind, asker string }{
{got, "go", "delivery", "builder"},
{got, "go", "delivery", "mesh-builds"},
{"mesh.seat.node-build-go.accept.check.mesh-pipeline-runner", "go", "check", "mesh-pipeline-runner"},
{"mesh.seat.node-build-node.accept.check", "node", "check", ""},
} {
l, k, a, err := ParseAskSubjectBy(c.subject)
modules/mesh-delivery-sdk/steprun/askers.go
+1 -1
View File
@@ -92,7 +92,7 @@ func (p *Puller) askerRefusal(msg *Pulled, ask sdkstep.StepAsk) (why string, jud
// refuseAsker ends an ask whose asker may not ask its kind: its step key says failed with the reason, naming the kind and the
// asker, step-ended says failed, and the message is dropped.
func (p *Puller) refuseAsker(msg *Pulled, ask sdkstep.StepAsk, why string) {
key, err := sdkstep.AskStepKey(p.Node, ask.ID)
key, err := sdkstep.AskStepKey(keyAsker(msg), p.Node, ask.ID)
if err == nil {
at := now().UTC()
step := sdkstep.Step{ID: ask.ID, Tool: ask.Kind + " ask", Node: p.Node, State: sdkstep.StepFailed, Started: at, Ended: at,
modules/mesh-delivery-sdk/steprun/askers_test.go
+10 -3
View File
@@ -39,7 +39,12 @@ func TestPullRefusesAKindItsAskerMayNotAsk(t *testing.T) {
t.Run(c.name, func(t *testing.T) {
src, st, sy := NewFakeAsks(), newSteps(), &sayer{}
raw, _ := sdkstep.EncodeStepAsk(c.ask)
src.PublishWithSubject(c.ask.Kind, raw, c.subject)
queue := c.ask.Kind
if queue == sdkstep.AskStep {
queue = sdkstep.AskCheck // a step rides the queue check (A-170, A-184)
}
src.PublishWithSubject(queue, raw, c.subject)
_, _, asker, _ := sdkstep.ParseAskSubjectBy(c.subject)
ran := make(chan struct{}, 1)
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 }, Callers: c.callers,
PullWait: 5 * time.Millisecond, Heartbeat: time.Hour, Say: sy.say, ProgressEvery: time.Hour,
@@ -50,11 +55,13 @@ func TestPullRefusesAKindItsAskerMayNotAsk(t *testing.T) {
switch c.want {
case "":
<-ran
// §481: the step is keyed under the asker its subject names, main's under its node alone.
key, _ := sdkstep.AskStepKey(asker, "shanks", c.ask.ID)
waitFor(t, "the step keyed "+key, func() bool { _, ok := st.get(key); return ok })
case "refused":
waitFor(t, "the ask ended", func() bool { _, termed, _ := src.Fates(); return len(termed) == 1 })
key, _ := sdkstep.AskStepKey("shanks", c.ask.ID)
key, _ := sdkstep.AskStepKey(asker, "shanks", c.ask.ID)
s, _ := st.get(key)
asker := c.subject[strings.LastIndex(c.subject, ".")+1:]
if s.State != sdkstep.StepFailed || !strings.Contains(s.Reason, KindOf(c.ask)) || (strings.Contains(c.subject, "mesh-catalog") && !strings.Contains(s.Reason, asker)) {
t.Fatalf("the step = %+v, want failed naming the kind %s and the asker", s, KindOf(c.ask))
}
modules/mesh-delivery-sdk/steprun/pull.go
+19 -4
View File
@@ -548,7 +548,7 @@ func (p *Puller) next(ctx context.Context, wait time.Duration) (*Pulled, error)
// message: the asker puts the ask again under a new id (issue 890). A message whose decline cannot be written is handed
// back, as one whose start cannot be.
func (p *Puller) decline(msg *Pulled, ask sdkstep.StepAsk, lacks []string) {
step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)))
step, key, err := beginRun(p.Put, keyAsker(msg), p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), 0)
if err == nil {
why := Refuse("the ask %s needs %s, and %s lacks %s: it is left to a node that has it", ask.ID,
strings.Join(ask.Needs, ", "), p.Node, strings.Join(lacks, ", "))
@@ -605,7 +605,7 @@ func (p *Puller) refuseUnread(msg *Pulled, why error) {
Kind string `json:"kind"`
}
if json.Unmarshal(msg.Data, &probe) == nil && probe.ID != "" {
if key, kerr := sdkstep.AskStepKey(p.Node, probe.ID); kerr == nil {
if key, kerr := sdkstep.AskStepKey(keyAsker(msg), p.Node, probe.ID); kerr == nil {
tool := "ask"
if probe.Kind != "" {
tool = probe.Kind + " ask"
@@ -637,7 +637,7 @@ func withKind(msg *Pulled, kind string) *Pulled {
func (p *Puller) take(ctx context.Context, msg *Pulled) {
kind := msg.Kind
ask, err := sdkstep.DecodeStepAsk(msg.Data)
if err == nil && (ask.Language != p.Language || (kind != "" && ask.Kind != kind)) {
if err == nil && (ask.Language != p.Language || (kind != "" && !ridesQueue(ask.Kind, kind))) {
err = fmt.Errorf("the ask %s is a %s ask of %s, pulled from the %s %s worker", ask.ID, ask.Kind, ask.Language, p.Language, kind)
}
if err != nil {
@@ -662,7 +662,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
p.decline(msg, ask, lacks)
return
}
step, key, err := beginRun(p.Put, p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), int(msg.Delivered))
step, key, err := beginRun(p.Put, keyAsker(msg), p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), int(msg.Delivered))
if err != nil {
p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
_ = msg.Nak()
@@ -954,6 +954,21 @@ type stepProgress struct {
At time.Time `json:"at"`
}
// keyAsker is the asker an ask's step is keyed under: the last token of the subject it came on, none for main's form or a
// source that gives no subject (design 47 G §481, §483), never the ask's own field.
func keyAsker(msg *Pulled) string {
if _, _, asker, err := sdkstep.ParseAskSubjectBy(msg.Subject); err == nil {
return asker
}
return ""
}
// ridesQueue says an ask of kind may come from the worker of queue: its own, or for a step check or delivery (design 47 G
// §415, §600, A-170).
func ridesQueue(kind, queue string) bool {
return kind == queue || (kind == sdkstep.AskStep && (queue == sdkstep.AskCheck || queue == sdkstep.AskDelivery))
}
// askerOf is the asker the bus proved: the last token of the subject the ask came on (design 47 G §481), the ask's own
// field only for a source that gives no subject and for main's form.
func askerOf(msg *Pulled, ask sdkstep.StepAsk) string {
modules/mesh-delivery-sdk/steprun/pull_drop.go
+1 -1
View File
@@ -134,7 +134,7 @@ func (p *Puller) droppedAsk(id string) *nodebuildgo.Drop {
// refuseDropped ends a dropped ask the bus gave again: its step refused naming who dropped it, the message ended and the
// drop's key deleted.
func (p *Puller) refuseDropped(msg *Pulled, ask sdkstep.StepAsk, d *nodebuildgo.Drop) {
key, err := sdkstep.AskStepKey(p.Node, ask.ID)
key, err := sdkstep.AskStepKey(keyAsker(msg), p.Node, ask.ID)
if err == nil {
at := now().UTC()
step := sdkstep.Step{ID: ask.ID, Tool: ask.Kind + " ask", Node: p.Node, State: sdkstep.StepDropped, Started: at, Ended: at, Owner: buildrun.Owner,
modules/mesh-delivery-sdk/steprun/pull_integration_test.go
+13 -5
View File
@@ -54,6 +54,14 @@ func publishAsk(t *testing.T, js jetstream.JetStream, a sdkstep.StepAsk) {
}
}
// askKey is the key of a's step on node as publishAsk put it: under its asker, main's form for a build (design 47 G §481).
func askKey(a sdkstep.StepAsk, node string) (string, error) {
if a.Kind == sdkstep.AskDelivery {
return sdkstep.AskStepKey("", node, a.ID)
}
return sdkstep.AskStepKey("mesh-delivery-checker", node, a.ID)
}
// 1887/A2a, 1887/AC-order, 1887/AC-pull on the real server: a check ask published before a delivery ask runs after it;
// with capacity 2 and three asks two run, the third waits in the stream and starts when one ends.
func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
@@ -90,7 +98,7 @@ func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
close(release)
waitFor(t, "every step ended", func() bool {
for _, a := range []sdkstep.StepAsk{check, first, second} {
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := askKey(a, "shanks")
if s, ok := st.get(key); !ok || s.State != sdkstep.StepPassed {
return false
}
@@ -134,7 +142,7 @@ func TestPullRedeliversAKilledPullersAsk(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go other.Pull(ctx)
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := askKey(a, "shanks")
deadline := time.Now().Add(10 * time.Second)
for {
if s, ok := st.get(key); ok && s.State == sdkstep.StepPassed {
@@ -183,7 +191,7 @@ func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("ace", a.ID)
key, _ := askKey(a, "ace")
waitFor(t, "the ask's second run ended", func() bool { s, ok := st.get(key); return ok && s.State == sdkstep.StepPassed })
mu.Lock()
gap := runs[1].Sub(runs[0])
@@ -196,7 +204,7 @@ func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) {
if err := p.Drop(doomed.ID, "mesh-delivery-builder", "superseded"); err != nil {
t.Fatal(err)
}
dkey, _ := sdkstep.AskStepKey("ace", doomed.ID)
dkey, _ := askKey(doomed, "ace")
waitFor(t, "the dropped step dropped", func() bool { s, ok := st.get(dkey); return ok && s.State == sdkstep.StepDropped })
if s, _ := st.get(dkey); !contains(s.Reason, "dropped by mesh-delivery-builder: superseded") {
t.Errorf("the dropped step says %q", s.Reason)
@@ -268,7 +276,7 @@ func TestACheckAnswerOverTheStepsStateIsKeptAndReadBackWhole(t *testing.T) {
KeepOutput: func(object string, body []byte) error { _, err := objects.PutBytes(ctx, object, body); return err },
Run: func(context.Context, sdkstep.StepAsk) (any, error) { return answer, nil }}
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("ace", a.ID)
key, _ := askKey(a, "ace")
var step sdkstep.Step
waitFor(t, "the check's step ended", func() bool {
e, err := kv.Get(ctx, key)
modules/mesh-delivery-sdk/steprun/pull_seat_test.go
+1 -1
View File
@@ -119,7 +119,7 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) {
}
waitFor(t, "the ask ended", func() bool { ack, tm, _ := src.Fates(); return len(ack)+len(tm) == 1 })
cancel()
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := sdkstep.AskStepKey("", "shanks", a.ID)
s, ok := st.get(key)
if !ok || s.State != sdkstep.StepDropped || !strings.Contains(s.Reason, "dropped by mesh-delivery-builder: superseded") || ran != running {
t.Errorf("running %v: step %+v (%v), ran %v", running, s, ok, ran)
modules/mesh-delivery-sdk/steprun/pull_test.go
+9 -9
View File
@@ -99,7 +99,7 @@ func TestPullTakesDeliveryBeforeCheck(t *testing.T) {
if want := []string{sdkstep.AskDelivery, sdkstep.AskCheck, sdkstep.AskRecheck}; len(order) != 3 || order[0] != want[0] || order[1] != want[1] || order[2] != want[2] {
t.Fatalf("order = %v, want %v", order, want)
}
key, _ := sdkstep.AskStepKey("shanks", deliver.ID)
key, _ := sdkstep.AskStepKey("", "shanks", deliver.ID)
s, ok := st.get(key)
if !ok || s.State != sdkstep.StepPassed || string(s.Result) != `{"ran":"`+deliver.ID+`"}` {
t.Fatalf("the step %s = %+v (%v)", key, s, ok)
@@ -167,7 +167,7 @@ func TestPullHoldsAtMostItsCapacity(t *testing.T) {
t.Fatalf("capacity.shanks = %s (%v), said %d times; want effective 2, said once", raw, err, capLines)
}
waitFor(t, "a phase on the step key", func() bool {
key, _ := sdkstep.AskStepKey("shanks", p.Running()[0].ID)
key, _ := sdkstep.AskStepKey("", "shanks", p.Running()[0].ID)
s, _ := st.get(key)
return s.State == sdkstep.StepRunning && s.Phase == "unit: tests of 3 packages" && !s.PhaseAt.IsZero() && s.Owner == buildrun.Owner // 1890/AC5
})
@@ -178,7 +178,7 @@ func TestPullHoldsAtMostItsCapacity(t *testing.T) {
close(release)
waitFor(t, "all acked", func() bool { a, _, _ := src.Fates(); return len(a) == 3 })
for _, a := range asks {
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := sdkstep.AskStepKey("", "shanks", a.ID)
if s, _ := st.get(key); s.State != sdkstep.StepPassed || s.Phase != "unit: tests of 3 packages" || s.Owner != buildrun.Owner {
t.Errorf("the end of %s = %+v, want passed in its last phase", a.ID, s)
}
@@ -219,7 +219,7 @@ func TestPullDropsWhatIsNoAsk(t *testing.T) {
if ran {
t.Fatal("a wrong or a declined ask ran")
}
key, _ := sdkstep.AskStepKey("ace", placed.ID)
key, _ := sdkstep.AskStepKey("", "ace", placed.ID)
s, _ := st.get(key)
if d, ok := sdkstep.DeclinedOf(s); !ok || len(d.Lacks) != 1 || d.Lacks[0] != sdkstep.NeedNodeBuildNode || !strings.Contains(s.Reason, "lacks node-build-node") {
t.Fatalf("the declined step = %+v", s)
@@ -241,7 +241,7 @@ func TestPullDropsWhatIsNoAsk(t *testing.T) {
}}
go s2.Pull(ctx)
waitFor(t, "the ask termed on the node that sits out", func() bool { _, tm, _ := sit.Fates(); return len(tm) == 1 })
sk, _ := sdkstep.AskStepKey("ace", toolchain.ID)
sk, _ := sdkstep.AskStepKey("", "ace", toolchain.ID)
ss, _ := sitSteps.get(sk)
if d, ok := sdkstep.DeclinedOf(ss); !ok || len(d.Lacks) != 1 || !strings.Contains(ss.Reason, "26.12.1") {
t.Fatalf("the sitting-out step = %+v", ss)
@@ -490,7 +490,7 @@ func TestPullHeartbeatAndStop(t *testing.T) {
if len(acked)+len(termed)+len(naked) != 0 {
t.Fatalf("a stopped holder gave its ask a fate: acked %v termed %v naked %v", acked, termed, naked)
}
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := sdkstep.AskStepKey("", "shanks", a.ID)
if s, ok := st.get(key); !ok || s.State != sdkstep.StepRunning {
t.Fatalf("the step %s = %+v, want left running", key, s)
}
@@ -510,7 +510,7 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
defer cancel()
go p.Pull(ctx)
waitFor(t, "the ask termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 1 })
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := sdkstep.AskStepKey("", "shanks", a.ID)
s, ok := st.get(key)
if ran || !ok || s.State != sdkstep.StepFailed || !strings.Contains(s.Reason, "taken 3 times") || !strings.Contains(s.Reason, "poison") {
t.Fatalf("ran %v, step %+v", ran, s)
@@ -543,7 +543,7 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { return nil, c.err }}
ctx, cancel := context.WithCancel(context.Background())
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("shanks", a.ID)
key, _ := sdkstep.AskStepKey("", "shanks", a.ID)
waitFor(t, c.name, func() bool {
ack, _, _ := src.Fates()
src.mu.Lock()
@@ -592,7 +592,7 @@ func TestPullAnswersAnAskItCannotRead(t *testing.T) {
defer cancel()
go p.Pull(ctx)
waitFor(t, "both termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 2 })
key, _ := sdkstep.AskStepKey("ace", good.ID)
key, _ := sdkstep.AskStepKey("", "ace", good.ID)
s, ok := st.get(key)
if !ok || s.State != sdkstep.StepRefused || !strings.Contains(s.Reason, "cannot read the ask") || !strings.Contains(s.Reason, "exactly one") {
t.Fatalf("the unread ask's step = %+v (%v)", s, ok)
modules/mesh-delivery-sdk/steprun/step.go
+5 -4
View File
@@ -49,13 +49,14 @@ func Start(ctx context.Context, put Put, node, tool string, run func(ctx context
// begin writes the step running under id, owned by this process (Owner).
func begin(put Put, node, tool, id, what string) (sdkstep.Step, string, error) {
return beginRun(put, node, tool, id, what, 0)
return beginRun(put, "", node, tool, id, what, 0)
}
// beginRun is begin for the run-th run of an ask (the bus's delivery count; 0 for a step that is no ask).
func beginRun(put Put, node, tool, id, what string, run int) (sdkstep.Step, string, error) {
// beginRun is begin for the run-th run of an ask (the bus's delivery count; 0 for a step that is no ask), keyed under the
// asker its subject names, or main's form for none (design 47 G §481, §483).
func beginRun(put Put, asker, node, tool, id, what string, run int) (sdkstep.Step, string, error) {
step := sdkstep.Step{ID: id, Tool: tool, Node: node, State: sdkstep.StepRunning, Started: now().UTC(), What: what, Owner: buildrun.Owner, Run: run}
key, err := sdkstep.StepKey(node, step.ID)
key, err := sdkstep.StepKeyOf(asker, node, step.ID)
if err != nil {
return sdkstep.Step{}, "", err
}
modules/mesh-grants/internal/buscompose/testdata/user-lists.json
Vendored
+1 -1
View File
File diff suppressed because one or more lines are too long
modules/mesh-grants/internal/grants/named_asks_test.go
+7
View File
@@ -4,6 +4,7 @@ import (
"slices"
"testing"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
@@ -33,4 +34,10 @@ func TestAcceptCallersOf(t *testing.T) {
if !slices.Equal(q["check"], []string{"mesh-catalog", "mesh-pipeline-runner"}) || !slices.Equal(q["delivery"], []string{"mesh-builds"}) || len(q) != 2 {
t.Errorf("queues: %v", q)
}
// A-170, A-184: the seat's workers are per queue, never per kind: step and run ride check or delivery.
build := catalogue.SeatFromDefinition(seatdef.Definition{Seat: "node-build-go", PulledByKind: true, Accepts: []seatdef.Message{
msg("delivery", ""), msg("check", ""), msg("step", "check"), msg("step", "delivery"), msg("recheck", ""), msg("run", "check")}})
if got := queuesOf(build); !slices.Equal(got, []string{"delivery", "check", "recheck"}) {
t.Errorf("the queues of %v on %v: %v, want delivery, check and recheck", build.Accepts, build.AcceptedOn, got)
}
}
modules/mesh-grants/internal/grants/records.go
+29
View File
@@ -453,11 +453,40 @@ func markedNamed(d buscompose.Declared, seats map[string]catalogue.SeatDeclarati
return slices.Compact(out)
}
// queuesOf are a pulled-by-kind seat's queues, in the order of its accepted kinds: a kind's stream, or its own name where it
// rides none (step and run ride check or delivery, so a build seat's are delivery, check and recheck). Each is one subject
// and one worker of each form (design 47 G §600, §544, A-170, A-184).
func queuesOf(own catalogue.Seat) []string {
streamed := map[string]bool{}
var out []string
for _, on := range own.AcceptedOn {
if stream, kind, ok := strings.Cut(on, "."); ok {
streamed[kind] = true
if !slices.Contains(out, stream) {
out = append(out, stream)
}
}
}
var queues []string
for _, a := range own.Accepts {
if !streamed[a] && !slices.Contains(queues, a) {
queues = append(queues, a)
}
}
for _, q := range out {
if !slices.Contains(queues, q) {
queues = append(queues, q)
}
}
return queues
}
func asSeat(s catalogue.SeatDeclaration, declarer string) buscompose.Seat {
seat := buscompose.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits,
Serves: catalogue.VerbNames(s.Serves), Kinded: s.Kinded, ByCaller: s.ByCaller, Proofs: s.Proofs, DeclaredBy: declarer}
if own, ok := catalogue.SeatNamed(s.Name); ok && own.PulledByKind {
seat.PulledByKind = true
seat.Accepts = queuesOf(own)
}
for _, r := range s.Records {
if declarer != "" {