From 7033d5e80d1f13f5473b6b3cfd9c7d9b13ceddbf Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:42:00 +0100 Subject: [PATCH] BG2, BG3: the puller refuses a kind its asker may not ask and pulls the named worker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An ask's asker is the last token of its accept subject; the holder checks the kind against the callers the build seat's definition lists (Puller.Callers, CallersFromDefinitions), refuses an ask of another kind answering error naming the kind and the asker and ending it failed, takes only a build and a run on main's subject, and gives back an ask the seat list does not judge yet. A second worker per queue (AskNamedWorker, filtered to accept..*) is pulled after main's; the holder says `named worker not bound` while its membership lists none. Builds: design 47 G §543, §544, §481, A-163, A-170 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- modules/mesh-delivery-sdk/buildstep/ask.go | 10 ++ modules/mesh-delivery-sdk/steprun/askers.go | 149 ++++++++++++++++++ .../mesh-delivery-sdk/steprun/askers_test.go | 86 ++++++++++ modules/mesh-delivery-sdk/steprun/pull.go | 12 +- modules/mesh-delivery-sdk/steprun/pull_bus.go | 78 +++++++-- .../mesh-delivery-sdk/steprun/pull_fake.go | 20 ++- .../steprun/pull_integration_test.go | 44 +++--- .../mesh-delivery-sdk/steprun/pull_seat.go | 52 +++++- .../steprun/pull_seat_test.go | 35 +++- 9 files changed, 449 insertions(+), 37 deletions(-) create mode 100644 modules/mesh-delivery-sdk/steprun/askers.go create mode 100644 modules/mesh-delivery-sdk/steprun/askers_test.go diff --git a/modules/mesh-delivery-sdk/buildstep/ask.go b/modules/mesh-delivery-sdk/buildstep/ask.go index 64439a89..e78e5c60 100644 --- a/modules/mesh-delivery-sdk/buildstep/ask.go +++ b/modules/mesh-delivery-sdk/buildstep/ask.go @@ -203,6 +203,16 @@ func AskWorker(language, kind string) (string, error) { return stream + "_" + strings.ToUpper(kind) + "_worker", nil } +// AskNamedWorker is the pull worker of a queue's named asks, `SEAT___named_worker`, filtered to +// `mesh.seat..accept..*`, beside main's worker AskWorker, filtered to main's subject alone (design 47 G §544). +func AskNamedWorker(language, kind string) (string, error) { + worker, err := AskWorker(language, kind) + if err != nil { + return "", err + } + 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) { if err := CheckAskID(askID); err != nil { diff --git a/modules/mesh-delivery-sdk/steprun/askers.go b/modules/mesh-delivery-sdk/steprun/askers.go new file mode 100644 index 00000000..3023ad38 --- /dev/null +++ b/modules/mesh-delivery-sdk/steprun/askers.go @@ -0,0 +1,149 @@ +package steprun + +import ( + "encoding/json" + "fmt" + "slices" + "strings" + "sync/atomic" + + buildrun "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrun" + sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/seatlist" +) + +// **A holder refuses a kind its asker may not ask** (design 47 G §543, A-163, A-170). An ask rides +// `mesh.seat..accept..`, the asker the asking seat's own name, and its kind stays in its body. The +// holder takes the asker from the subject and checks the kind against the callers the seat's definition lists for it (read +// from the seat list); on main's subject, which names no asker, it takes only a build and a run. + +// KindOf is the word the seat definition gives an ask's kind: run for a run or a compose, else the ask's own kind. +func KindOf(ask sdkstep.StepAsk) string { + if ask.Run != nil || ask.Compose != nil { + return "run" + } + return ask.Kind +} + +// CallersFromDefinitions is a Puller.Callers: the seats the definition of language's build seat lists as callers of an +// accepted kind, from the seat list's definitions (seatlist.Reader.Definitions). found is false when the definition or the +// kind is not held yet. A caller entry is `seat:`. +func CallersFromDefinitions(language string, definitions func() map[string]seatdef.Definition) func(kind string) (callers []string, found bool) { + return func(kind string) ([]string, bool) { + seat, err := sdkstep.BuildSeat(language) + if err != nil { + return nil, false + } + def, ok := definitions()[seat] + if !ok { + return nil, false + } + var out []string + found := false + for _, m := range def.Accepts { + if m.Name != kind { + continue + } + found = true + if m.Callers != nil { + for _, e := range m.Callers.List { + if name, ok := strings.CutPrefix(e, "seat:"); ok { + out = append(out, name) + } + } + } + } + return out, found + } +} + +// askerRefusal is why an ask must not be taken from its subject's asker: "" when it may be, and judged false when the seat +// list does not say yet (the ask is given back, not refused). +func (p *Puller) askerRefusal(msg *Pulled, ask sdkstep.StepAsk) (why string, judged bool) { + if msg.Subject == "" { + return "", true // a source that gives no subject (a test's) + } + _, _, asker, err := sdkstep.ParseAskSubjectBy(msg.Subject) + if err != nil { + return fmt.Sprintf("the ask came on %s, which is no accept subject of this seat", msg.Subject), true + } + kind := KindOf(ask) + if asker == "" { + if (ask.Build != nil && ask.Kind == sdkstep.AskDelivery) || kind == "run" { + return "", true + } + return fmt.Sprintf("the kind %s is not taken on main's subject, which takes only a build and a run (asker: none named)", kind), true + } + if p.Callers == nil { + return "", true + } + callers, found := p.Callers(kind) + if !found { + return "", false + } + if !slices.Contains(callers, asker) { + return fmt.Sprintf("the kind %s is not one %s may ask: its callers are %s", kind, asker, strings.Join(callers, ", ")), true + } + return "", true +} + +// 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) + 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, + Reason: "error: " + why, Owner: buildrun.Owner, Run: int(msg.Delivered)} + err = writeEnd(p.Put, key, step) + } + if err != nil { + p.logf("the ask %s of the wrong asker cannot be answered on %s, so it is given back: %v", ask.ID, p.Node, err) + _ = msg.Nak() + return + } + p.logf("the ask %s is refused on %s: %s", ask.ID, p.Node, why) + if ask.Step == nil { + p.sayEnded(msg, ask, sdkstep.OutcomeFailed, why) + } else if end, err := sdkstep.EndOf(ask, p.Node, sdkstep.OutcomeFailed); err == nil { + // A pipeline step's end names its pipeline, step, commit and fingerprint (its own form, §602). + end.Said, end.At, end.Asker = why, now().UTC(), askerOf(msg, ask) + if body, err := sdkstep.EncodeStepEnded(end); err == nil { + p.say("step-ended", json.RawMessage(body), "step-ended."+ask.ID) + } + } + _ = msg.Term() +} + +// giveBack hands an ask back to be judged again a little later, once the seat list says more. +func (p *Puller) giveBack(msg *Pulled, ask sdkstep.StepAsk) { + p.logf("the ask %s is given back on %s: the seat list holds no definition of its kind yet", ask.ID, p.Node) + if msg.NakAfter != nil && msg.NakAfter(DefaultPullWait) == nil { + return + } + _ = msg.Nak() +} + +// SeatListCallers is a Puller.Callers that follows the seat list: it starts the reader once the runtime serves +// (seatlist.Start through the runtime's own way in) and answers from its definitions, found false until it has read them. +// A reader that cannot start is said and the asks are given back until it can (design 47 G §543, A-163). +func SeatListCallers(language string, logf func(format string, args ...any)) func(kind string) ([]string, bool) { + var reader atomic.Pointer[seatlist.Reader] + go func() { + <-stdio.Served() + r, err := seatlist.Start(seatlist.Config{Bus: seatlist.StdioBus(), Bootstrap: seatlist.Bootstrapped()}) + if err != nil && logf != nil { + logf("following the seat list: %v", err) + } + reader.Store(r) + }() + return func(kind string) ([]string, bool) { + r := reader.Load() + if r == nil { + return nil, false + } + return CallersFromDefinitions(language, r.Definitions)(kind) + } +} diff --git a/modules/mesh-delivery-sdk/steprun/askers_test.go b/modules/mesh-delivery-sdk/steprun/askers_test.go new file mode 100644 index 00000000..aa5ace6e --- /dev/null +++ b/modules/mesh-delivery-sdk/steprun/askers_test.go @@ -0,0 +1,86 @@ +package steprun + +import ( + "context" + "strings" + "testing" + "time" + + sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-sdk/go/seatdef" +) + +func callersFor(table map[string][]string) func(string) ([]string, bool) { + return func(kind string) ([]string, bool) { c, ok := table[kind]; return c, ok } +} + +// Design 47 G §543 (BG2, BG3, A-163, A-170): an ask whose kind its asker, the last token of its subject, is no caller of is +// answered error naming the kind and the asker and ended failed; a step on main's subject is refused as well; one the +// seat list does not judge yet is given back. +func TestPullRefusesAKindItsAskerMayNotAsk(t *testing.T) { + step := testAsk(sdkstep.AskStep) + step.Check = nil + step.Step = &sdkstep.StepRun{Pipeline: "p", Step: sdkstep.StepUnit, Module: "m", Artifact: "m:a", Commit: "0123456789abcdef0123456789abcdef01234567", + Fingerprint: "fp", Fetch: sdkstep.Fetch{Repository: "novox/x", Commit: "0123456789abcdef0123456789abcdef01234567"}} + build := testAsk(sdkstep.AskDelivery) + table := map[string][]string{"step": {"mesh-pipeline-runner"}, "delivery": {"mesh-builds"}} + for _, c := range []struct { + name, subject string + ask sdkstep.StepAsk + callers func(string) ([]string, bool) + want string // "" runs it, "refused" refuses it, "back" gives it back + }{ + {"the indexer's step", "mesh.seat.node-build-go.accept.check.mesh-catalog", step, callersFor(table), "refused"}, + {"the runner's step", "mesh.seat.node-build-go.accept.check.mesh-pipeline-runner", step, callersFor(table), ""}, + {"a step on main's subject", "mesh.seat.node-build-go.accept.check", step, callersFor(table), "refused"}, + {"a build on main's subject", "mesh.seat.node-build-go.accept.delivery", build, callersFor(table), ""}, + {"a kind the seat list does not hold", "mesh.seat.node-build-go.accept.check.mesh-pipeline-runner", step, callersFor(nil), "back"}, + } { + 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) + 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, + Run: func(context.Context, sdkstep.StepAsk) (any, error) { ran <- struct{}{}; return nil, nil }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + switch c.want { + case "": + <-ran + case "refused": + waitFor(t, "the ask ended", func() bool { _, termed, _ := src.Fates(); return len(termed) == 1 }) + key, _ := sdkstep.AskStepKey("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)) + } + if end := sy.of("step-ended"); len(end) != 1 || end[0].body["outcome"] != sdkstep.OutcomeFailed { + t.Fatalf("step-ended = %+v, want failed", end) + } + select { + case <-ran: + t.Fatal("a refused ask ran") + default: + } + case "back": + waitFor(t, "the ask given back", func() bool { src.mu.Lock(); defer src.mu.Unlock(); return len(src.Delayed) > 0 }) + } + }) + } +} + +func TestCallersFromDefinitionsReadsTheSeatsCallers(t *testing.T) { + def := seatdef.Definition{Seat: "node-build-go", Accepts: []seatdef.Message{ + {Name: "step", Callers: &seatdef.Callers{List: []string{"seat:mesh-pipeline-runner"}}}}} + callers := CallersFromDefinitions(sdkstep.AskGo, func() map[string]seatdef.Definition { return map[string]seatdef.Definition{"node-build-go": def} }) + if got, ok := callers("step"); !ok || len(got) != 1 || got[0] != "mesh-pipeline-runner" { + t.Fatalf("callers of step = %v, %v", got, ok) + } + if _, ok := callers("run"); ok { + t.Fatal("a kind the definition does not list is found") + } +} diff --git a/modules/mesh-delivery-sdk/steprun/pull.go b/modules/mesh-delivery-sdk/steprun/pull.go index 26677b54..77a772d6 100644 --- a/modules/mesh-delivery-sdk/steprun/pull.go +++ b/modules/mesh-delivery-sdk/steprun/pull.go @@ -80,7 +80,10 @@ type Puller struct { // Say publishes one event of the seat on this node's subject (SayOn): ask-taken when an ask starts, step-progress once // every ProgressEvery while it runs, step-ended when an ask outside a pipeline ends (design 47 G §496, §602). Nil says // none, a test's. ProgressEvery defaults to DefaultProgressEvery. - Say func(event string, body any, id string) error + Say func(event string, body any, id string) error + // Callers says which seats the build seat's definition lists as callers of an accepted kind (CallersFromDefinitions); + // found is false while the seat list holds no such kind. Nil checks no named asker (design 47 G §543, A-163). + Callers func(kind string) (callers []string, found bool) ProgressEvery time.Duration Logf func(format string, args ...any) // Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull @@ -640,6 +643,13 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { p.refuseUnread(msg, err) return } + if why, judged := p.askerRefusal(msg, ask); !judged { + p.giveBack(msg, ask) + return + } else if why != "" { + p.refuseAsker(msg, ask, why) + return + } if d := p.droppedAsk(ask.ID); d != nil { p.refuseDropped(msg, ask, d) return diff --git a/modules/mesh-delivery-sdk/steprun/pull_bus.go b/modules/mesh-delivery-sdk/steprun/pull_bus.go index e261a1d0..5669ec4b 100644 --- a/modules/mesh-delivery-sdk/steprun/pull_bus.go +++ b/modules/mesh-delivery-sdk/steprun/pull_bus.go @@ -16,14 +16,19 @@ type JetStreamAsks struct { JS jetstream.JetStream Language string - mu sync.Mutex - cons map[string]jetstream.Consumer + mu sync.Mutex + cons map[string]jetstream.Consumer + missing map[string]bool } -func (s *JetStreamAsks) consumer(ctx context.Context, kind string) (jetstream.Consumer, error) { +func (s *JetStreamAsks) consumer(ctx context.Context, kind string, named bool) (jetstream.Consumer, error) { s.mu.Lock() defer s.mu.Unlock() - if c, ok := s.cons[kind]; ok { + cacheKey := kind + if named { + cacheKey = kind + "/named" + } + if c, ok := s.cons[cacheKey]; ok { return c, nil } stream, err := sdkstep.AskStream(s.Language) @@ -31,6 +36,9 @@ func (s *JetStreamAsks) consumer(ctx context.Context, kind string) (jetstream.Co return nil, err } worker, err := sdkstep.AskWorker(s.Language, kind) + if named { + worker, err = sdkstep.AskNamedWorker(s.Language, kind) + } if err != nil { return nil, err } @@ -41,17 +49,60 @@ func (s *JetStreamAsks) consumer(ctx context.Context, kind string) (jetstream.Co if s.cons == nil { s.cons = map[string]jetstream.Consumer{} } - s.cons[kind] = c + s.cons[cacheKey] = c return c, nil } -// Next is AskSource's: a fetch of one with no wait when wait is 0, else one that waits at most wait. +// Next is AskSource's: main's worker first, then the queue's named worker once it is bound (design 47 G §544), with no +// wait when wait is 0, else waiting at most wait between the two. A named worker not bound is only main's alone. func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) { - c, err := s.consumer(ctx, kind) - if err != nil { - return nil, err + var half time.Duration + if wait > 0 { + half = wait / 2 } + for i, step := range []struct { + named bool + wait time.Duration + }{{false, 0}, {true, 0}, {false, half}, {true, wait - half}} { + if step.named && s.unbound(kind) { + continue + } + if i >= 2 && wait <= 0 { + break // the two waiting pulls are for a wait + } + c, err := s.consumer(ctx, kind, step.named) + if err != nil { + if step.named { + s.markUnbound(kind) + continue + } + return nil, err + } + if msg, err := s.fetch(ctx, c, step.wait); err != nil || msg != nil { + return msg, err + } + } + return nil, nil +} + +func (s *JetStreamAsks) unbound(kind string) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.missing[kind] +} + +func (s *JetStreamAsks) markUnbound(kind string) { + s.mu.Lock() + if s.missing == nil { + s.missing = map[string]bool{} + } + s.missing[kind] = true + s.mu.Unlock() +} + +func (s *JetStreamAsks) fetch(ctx context.Context, c jetstream.Consumer, wait time.Duration) (*Pulled, error) { var batch jetstream.MessageBatch + var err error if wait <= 0 { batch, err = c.FetchNoWait(1) } else { @@ -93,11 +144,16 @@ func DeclareAskWorkers(ctx context.Context, js jetstream.JetStream, language str for _, kind := range sdkstep.AskKinds { worker, _ := sdkstep.AskWorker(language, kind) subject, _ := sdkstep.AskSubject(language, kind) - // Both forms of the accept subject: main's and the asker's, `.` (design 47 G §600). - if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubjects: []string{subject, subject + ".*"}, + // Two workers whose filters do not overlap: main's subject and the asker's, `.` (design 47 G §544). + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubjects: []string{subject}, AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5}); err != nil { return fmt.Errorf("the worker %s: %w", worker, err) } + named, _ := sdkstep.AskNamedWorker(language, kind) + if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: named, FilterSubjects: []string{subject + ".*"}, + AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5}); err != nil { + return fmt.Errorf("the worker %s: %w", named, err) + } } return nil } diff --git a/modules/mesh-delivery-sdk/steprun/pull_fake.go b/modules/mesh-delivery-sdk/steprun/pull_fake.go index 574605dc..05e648e3 100644 --- a/modules/mesh-delivery-sdk/steprun/pull_fake.go +++ b/modules/mesh-delivery-sdk/steprun/pull_fake.go @@ -13,7 +13,9 @@ import ( type FakeAsks struct { mu sync.Mutex waiting map[string][][]byte - added chan struct{} + // subjects are the subjects asks were published with, by their data. + subjects map[string]string + added chan struct{} // Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given. Acked, Termed, Naked []string // Delayed are the asks handed back to be given again after a delay (NakAfter), and Delays that delay of each. @@ -40,6 +42,17 @@ func (f *FakeAsks) Publish(kind string, data []byte) { } } +// PublishWithSubject puts an encoded ask on kind's worker as the bus would hand it over from subject (design 47 G §481). +func (f *FakeAsks) PublishWithSubject(kind string, data []byte, subject string) { + f.mu.Lock() + if f.subjects == nil { + f.subjects = map[string]string{} + } + f.subjects[string(data)] = subject + f.mu.Unlock() + f.Publish(kind, data) +} + // PublishAsk encodes a and puts it on its kind's worker. func (f *FakeAsks) PublishAsk(a sdkstep.StepAsk) error { raw, err := sdkstep.EncodeStepAsk(a) @@ -107,7 +120,10 @@ func (f *FakeAsks) pulled(data []byte) *Pulled { return nil } } + f.mu.Lock() + subject := f.subjects[string(data)] delivered := f.Delivered + f.mu.Unlock() if delivered == 0 { delivered = 1 } @@ -117,6 +133,6 @@ func (f *FakeAsks) pulled(data []byte) *Pulled { f.mu.Unlock() return nil } - return &Pulled{Data: data, Delivered: delivered, NakAfter: nakAfter, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), + return &Pulled{Data: data, Subject: subject, Delivered: delivered, NakAfter: nakAfter, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }} } diff --git a/modules/mesh-delivery-sdk/steprun/pull_integration_test.go b/modules/mesh-delivery-sdk/steprun/pull_integration_test.go index 42575013..07336764 100644 --- a/modules/mesh-delivery-sdk/steprun/pull_integration_test.go +++ b/modules/mesh-delivery-sdk/steprun/pull_integration_test.go @@ -40,7 +40,11 @@ func publishAsk(t *testing.T, js jetstream.JetStream, a sdkstep.StepAsk) { if err != nil { t.Fatal(err) } + // A build rides main's subject; any other kind a named asker's (design 47 G §543: main takes only a build and a run). subject, _ := sdkstep.AskSubject(a.Language, a.Kind) + if a.Kind != sdkstep.AskDelivery { + subject, _ = sdkstep.AskSubjectBy(a.Language, a.Kind, "mesh-delivery-checker") + } if _, err := js.Publish(context.Background(), subject, raw); err != nil { t.Fatal(err) } @@ -197,31 +201,35 @@ func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) { func contains(s, sub string) bool { return strings.Contains(s, sub) } -// Design 47 G §600: the seat's worker takes an ask put on either form of the accept subject, main's and the asker's. -func TestTheWorkerTakesBothFormsOfTheAcceptSubject(t *testing.T) { +// Design 47 G §544: main's worker takes an ask put on main's subject and the named worker one put on the asker's, the two +// filters not overlapping. +func TestTheTwoWorkersTakeTheTwoFormsOfTheAcceptSubject(t *testing.T) { js := askBus(t, 30*time.Second) stream, _ := sdkstep.AskStream(sdkstep.AskGo) worker, _ := sdkstep.AskWorker(sdkstep.AskGo, sdkstep.AskDelivery) + named, _ := sdkstep.AskNamedWorker(sdkstep.AskGo, sdkstep.AskDelivery) main, _ := sdkstep.AskSubject(sdkstep.AskGo, sdkstep.AskDelivery) - for _, subject := range []string{main, main + ".builder"} { + for _, subject := range []string{main, main + ".mesh-builds"} { if _, err := js.Publish(context.Background(), subject, []byte("{}")); err != nil { t.Fatal(err) } } - c, err := js.Consumer(context.Background(), stream, worker) - if err != nil { - t.Fatal(err) - } - batch, err := c.Fetch(2, jetstream.FetchMaxWait(3*time.Second)) - if err != nil { - t.Fatal(err) - } - var got []string - for m := range batch.Messages() { - got = append(got, m.Subject()) - _ = m.Ack() - } - if len(got) != 2 { - t.Errorf("the worker took %v, want both forms of %s", got, main) + for name, want := range map[string]string{worker: main, named: main + ".mesh-builds"} { + c, err := js.Consumer(context.Background(), stream, name) + if err != nil { + t.Fatal(err) + } + batch, err := c.Fetch(2, jetstream.FetchMaxWait(time.Second)) + if err != nil { + t.Fatal(err) + } + var got []string + for m := range batch.Messages() { + got = append(got, m.Subject()) + _ = m.Ack() + } + if len(got) != 1 || got[0] != want { + t.Errorf("%s took %v, want only %s", name, got, want) + } } } diff --git a/modules/mesh-delivery-sdk/steprun/pull_seat.go b/modules/mesh-delivery-sdk/steprun/pull_seat.go index 91e5c99b..9958e5a5 100644 --- a/modules/mesh-delivery-sdk/steprun/pull_seat.go +++ b/modules/mesh-delivery-sdk/steprun/pull_seat.go @@ -2,6 +2,7 @@ package steprun import ( "context" + "sync" "time" sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" @@ -18,10 +19,15 @@ type SeatAsks struct { Settle func(id, fate string) error // SettleAfter is stdio.SeatSettleAfter when nil (ADR 0664 §1). SettleAfter func(id string, delay time.Duration) error + // Logf says `named worker not bound` once for each queue whose named worker the membership does not list yet. + Logf func(format string, args ...any) + + mu sync.Mutex + noted map[string]bool } // Next is AskSource's. -func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) { +func (s *SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*Pulled, error) { pull, settle := s.Pull, s.Settle if pull == nil { pull = stdio.SeatPull @@ -36,7 +42,33 @@ func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*P if err := ctx.Err(); err != nil { return nil, err } - p, err := pull(worker, wait) + named, err := sdkstep.AskNamedWorker(s.Language, kind) + if err != nil { + return nil, err + } + // Main's worker first, then the named one (design 47 G §544): a holder of G pulls main's alone until its membership + // names the named worker, and says so meanwhile. A wait is shared by the two. + half := wait / 2 + p, err := pull(worker, min(wait, 0)) + if err == nil && p == nil { + var nerr error + if p, nerr = pull(named, 0); nerr != nil { + s.noteUnbound(kind, nerr) + p = nil + } else { + s.mu.Lock() + delete(s.noted, kind) + s.mu.Unlock() + } + } + if err == nil && p == nil && wait > 0 { + if p, err = pull(worker, half); err == nil && p == nil { + if p, err = pull(named, wait-half); err != nil { + s.noteUnbound(kind, err) + p, err = nil, nil + } + } + } if err != nil || p == nil { return nil, err } @@ -45,7 +77,21 @@ func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*P after = stdio.SeatSettleAfter } fate := func(f string) func() error { return func() error { return settle(p.ID, f) } } - return &Pulled{Data: p.Body, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress), + return &Pulled{Data: p.Body, Subject: p.Subject, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress), Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm), NakAfter: func(delay time.Duration) error { return after(p.ID, delay) }}, nil } + +// noteUnbound says `named worker not bound` for a queue, once until it is bound (design 47 G §544). +func (s *SeatAsks) noteUnbound(kind string, err error) { + s.mu.Lock() + if s.noted == nil { + s.noted = map[string]bool{} + } + said := s.noted[kind] + s.noted[kind] = true + s.mu.Unlock() + if !said && s.Logf != nil { + s.Logf("named worker not bound: the %s queue's named asks wait until the membership lists %s: %v", kind, "its named worker", err) + } +} diff --git a/modules/mesh-delivery-sdk/steprun/pull_seat_test.go b/modules/mesh-delivery-sdk/steprun/pull_seat_test.go index 2a4d64e6..0750b807 100644 --- a/modules/mesh-delivery-sdk/steprun/pull_seat_test.go +++ b/modules/mesh-delivery-sdk/steprun/pull_seat_test.go @@ -18,7 +18,7 @@ import ( func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) { var pulled []string var settled []string - src := SeatAsks{Language: sdkstep.AskGo, + src := &SeatAsks{Language: sdkstep.AskGo, Pull: func(worker string, wait time.Duration) (*stdio.Pulled, error) { pulled = append(pulled, worker) if worker != "SEAT_NODE_BUILD_GO_CHECK_worker" { @@ -36,7 +36,7 @@ func TestSeatAsksPullsTheKindsWorkerAndSettlesByID(t *testing.T) { } _ = p.InProgress() _ = p.Ack() - if len(pulled) != 2 || pulled[0] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" { + if len(pulled) != 3 || pulled[0] != "SEAT_NODE_BUILD_GO_DELIVERY_worker" || pulled[1] != "SEAT_NODE_BUILD_GO_DELIVERY_named_worker" { t.Errorf("pulled %v", pulled) } if len(settled) != 2 || settled[0] != "p9 progress" || settled[1] != "p9 ack" { @@ -146,3 +146,34 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) { } func ptr[T any](v T) *T { return &v } + +// Design 47 G §544 (BG3): a holder pulls main's worker, then the queue's named worker; while its membership does not list the +// named worker it takes main's alone and says `named worker not bound` once. +func TestSeatAsksPullsTheNamedWorkerAndSaysWhenItIsNotBound(t *testing.T) { + var said []string + bound := false + src := &SeatAsks{Language: sdkstep.AskGo, Logf: func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }, + Pull: func(worker string, wait time.Duration) (*stdio.Pulled, error) { + if worker != "SEAT_NODE_BUILD_GO_CHECK_named_worker" { + return nil, nil + } + if !bound { + return nil, fmt.Errorf("the module pulls no work from %s", worker) + } + return &stdio.Pulled{ID: "n1", Worker: worker, Subject: "mesh.seat.node-build-go.accept.check.mesh-pipeline-runner", Body: []byte(`{"v":1}`), Delivered: 1}, nil + }, + Settle: func(string, string) error { return nil }} + for range 2 { + if p, err := src.Next(context.Background(), sdkstep.AskCheck, 0); err != nil || p != nil { + t.Fatalf("unbound: %+v, %v", p, err) + } + } + if len(said) != 1 || !strings.Contains(said[0], "named worker not bound") { + t.Fatalf("said %q, want `named worker not bound` once", said) + } + bound = true + p, err := src.Next(context.Background(), sdkstep.AskCheck, 0) + if err != nil || p == nil || !strings.HasSuffix(p.Subject, ".mesh-pipeline-runner") { + t.Fatalf("bound: %+v, %v", p, err) + } +}