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) + } +}