diff --git a/modules/mesh-delivery-sdk/steprun/phase.go b/modules/mesh-delivery-sdk/steprun/phase.go index a4e878a7..72d38465 100644 --- a/modules/mesh-delivery-sdk/steprun/phase.go +++ b/modules/mesh-delivery-sdk/steprun/phase.go @@ -65,3 +65,10 @@ func (r *phases) end() sdkstep.Step { r.ended = true return r.step } + +// phase is the phase the ask is in now, "" before its first. +func (r *phases) phase() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.step.Phase +} diff --git a/modules/mesh-delivery-sdk/steprun/pull.go b/modules/mesh-delivery-sdk/steprun/pull.go index c3599866..26677b54 100644 --- a/modules/mesh-delivery-sdk/steprun/pull.go +++ b/modules/mesh-delivery-sdk/steprun/pull.go @@ -27,6 +27,9 @@ type Pulled struct { // Kind is the worker it came from, filled by the puller. Kind string Data []byte + // Subject is the subject the ask was published on: its last token is the asker the bus proved (design 47 G §481); empty + // for a source that has none, and for main's form, which names none. + Subject string // Delivered is how many times the bus has handed it over, this time included. Delivered uint64 // Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its @@ -53,6 +56,8 @@ var ( // DefaultHeartbeat is how often a running ask tells the bus it is still being worked; under the seat worker's // acknowledgement wait of 60 s (mesh-controller internal/broker HolderConsumerFor). DefaultHeartbeat = 20 * time.Second + // DefaultProgressEvery is how often a running ask says step-progress (design 47 G §496: once a minute). + DefaultProgressEvery = time.Minute ) // Puller pulls a build seat's asks and runs them. @@ -72,7 +77,12 @@ type Puller struct { // sdkstep.AgainAfter (a minute, then four). A test's own is short. Delay func(run int) time.Duration Heartbeat time.Duration - Logf func(format string, args ...any) + // 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 + 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 // reads the node's pause key (sdkstep.PauseKey) before it pulls, and Pause and Resume write it with PausePut, so a // pause outlives a restart. @@ -665,6 +675,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { p.running[ask.ID] = RunningAsk{ID: ask.ID, Kind: ask.Kind, Module: module, Started: step.Started} p.mu.Unlock() p.logf("the %s ask %s starts on %s: %s", ask.Kind, ask.ID, p.Node, step.What) + p.say("ask-taken", askTaken{Ask: ask.ID, Asker: askerOf(msg, ask), Node: p.Node, At: now().UTC()}, "ask-taken."+ask.ID) go p.run(ctx, msg, ask, key, step) } @@ -696,6 +707,25 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key } }() ph := &phases{put: p.Put, key: key, step: step, logf: p.Logf, prefix: fmt.Sprintf("the %s ask %s on %s", ask.Kind, ask.ID, p.Node)} + if p.Say != nil { + every := p.ProgressEvery + if every <= 0 { + every = DefaultProgressEvery + } + go func() { + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-done: + return + case <-t.C: + p.say("step-progress", stepProgress{Ask: ask.ID, Asker: askerOf(msg, ask), Phase: progressPhase(ask, ph.phase()), At: now().UTC()}, + fmt.Sprintf("step-progress.%s.%d", ask.ID, now().UnixNano())) + } + } + }() + } runCtx := p.cancellable(ctx, ask.ID) result, err := protect(runCtx, func(ctx context.Context) (any, error) { return p.Run(context.WithValue(ctx, phaseKey{}, ph), ask) @@ -713,6 +743,7 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key var dec *Declining if errors.As(err, &dec) && ctx.Err() == nil { p.declineRun(msg, ask, key, step, dec) + p.sayEnded(msg, ask, sdkstep.OutcomeCouldNotRun, dec.Text) return } if ctx.Err() != nil { @@ -750,6 +781,9 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key if len(outcome) > 300 { outcome = outcome[:300] + "..." } + p.sayEnded(msg, ask, sdkstep.OutcomeFailed, err.Error()) + } else { + p.sayEnded(msg, ask, sdkstep.OutcomePassed, "") } p.logf("the %s ask %s ended on %s after %s, in its phase %q; %s", ask.Kind, ask.ID, p.Node, now().Sub(step.Started).Round(time.Second), step.Phase, outcome) @@ -884,3 +918,70 @@ func askRead(ask sdkstep.StepAsk) sdkstep.Ask { out, _ := sdkstep.AskAs[sdkstep.Ask](ask) return out } + +// askTaken and stepProgress are the bodies of the seat's events ask-taken and step-progress (design 47 G §602, §496). +type askTaken struct { + Ask string `json:"ask"` + Asker string `json:"asker"` + Node string `json:"node"` + At time.Time `json:"at"` +} + +type stepProgress struct { + Ask string `json:"ask"` + Asker string `json:"asker"` + Phase string `json:"phase"` + At time.Time `json:"at"` +} + +// 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 { + if _, _, asker, err := sdkstep.ParseAskSubjectBy(msg.Subject); err == nil && asker != "" { + return asker + } + return ask.Asker +} + +// progressPhase is the phase step-progress says (fetch, build, run or report): fetch until the ask has entered a phase of +// its own or while it names the fetch, then the ask's kind (build for a build, run for any other). +func progressPhase(ask sdkstep.StepAsk, phase string) string { + switch { + case phase == "" || strings.HasPrefix(phase, "fetch"): + return "fetch" + case ask.Build != nil: + return "build" + } + return "run" +} + +// say publishes one event of the seat; a failure is logged and never fails the ask. +func (p *Puller) say(event string, body any, id string) { + if p.Say == nil { + return + } + if err := p.Say(event, body, id); err != nil { + p.logf("the event %s of the ask %s could not be said on %s: %v", event, id, p.Node, err) + } +} + +// sayEnded says step-ended for an ask outside a pipeline, whatever its asker (design 47 G §602): a pipeline step's end is +// its run's own (AskSteps.SayStep). +func (p *Puller) sayEnded(msg *Pulled, ask sdkstep.StepAsk, outcome, said string) { + if p.Say == nil || ask.Step != nil { + return + } + if len(said) > 300 { + said = said[:300] + "..." + } + if said == "" && outcome != sdkstep.OutcomePassed { + said = "the ask did not pass" + } + end := sdkstep.StepEnded{ID: ask.ID, Asker: askerOf(msg, ask), Ask: ask.ID, Node: p.Node, Outcome: outcome, Said: said, At: now().UTC()} + body, err := sdkstep.EncodeStepEnded(end) + if err != nil { + p.logf("the end of the ask %s cannot be said on %s: %v", ask.ID, p.Node, err) + return + } + p.say("step-ended", json.RawMessage(body), "step-ended."+ask.ID) +} diff --git a/modules/mesh-delivery-sdk/steprun/pull_bus.go b/modules/mesh-delivery-sdk/steprun/pull_bus.go index efedb295..e261a1d0 100644 --- a/modules/mesh-delivery-sdk/steprun/pull_bus.go +++ b/modules/mesh-delivery-sdk/steprun/pull_bus.go @@ -69,7 +69,7 @@ func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duratio if meta, err := msg.Metadata(); err == nil && meta != nil { delivered = meta.NumDelivered } - return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered, + return &Pulled{Data: append([]byte(nil), msg.Data()...), Subject: msg.Subject(), Delivered: delivered, Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term, NakAfter: func(delay time.Duration) error { return msg.NakWithDelay(delay) }}, nil case <-ctx.Done(): diff --git a/modules/mesh-delivery-sdk/steprun/pull_events_test.go b/modules/mesh-delivery-sdk/steprun/pull_events_test.go new file mode 100644 index 00000000..466cc951 --- /dev/null +++ b/modules/mesh-delivery-sdk/steprun/pull_events_test.go @@ -0,0 +1,93 @@ +package steprun + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + sdkstep "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" +) + +type said struct { + event, id string + body map[string]any +} + +type sayer struct { + mu sync.Mutex + got []said +} + +func (s *sayer) say(event string, body any, id string) error { + raw, _ := json.Marshal(body) + var m map[string]any + _ = json.Unmarshal(raw, &m) + s.mu.Lock() + s.got = append(s.got, said{event, id, m}) + s.mu.Unlock() + return nil +} + +func (s *sayer) of(event string) []said { + s.mu.Lock() + defer s.mu.Unlock() + var out []said + for _, g := range s.got { + if g.event == event { + out = append(out, g) + } + } + return out +} + +// Design 47 G §496, §602 (BG4, BN4, BI4): a holder says ask-taken when an ask starts, step-progress while it runs and +// step-ended when an ask outside a pipeline ends, whatever its asker. +func TestPullSaysAskTakenProgressAndEnd(t *testing.T) { + for _, c := range []struct { + name string + runErr error + outcome string + }{{"passed", nil, sdkstep.OutcomePassed}, {"failed", errors.New("the build broke"), sdkstep.OutcomeFailed}} { + t.Run(c.name, func(t *testing.T) { + src, st, sy := NewFakeAsks(), newSteps(), &sayer{} + a := testAsk(sdkstep.AskDelivery) + _ = src.PublishAsk(a) + release, entered := make(chan struct{}), make(chan struct{}) + p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 }, + PullWait: 10 * time.Millisecond, Heartbeat: time.Hour, Say: sy.say, ProgressEvery: 5 * time.Millisecond, + Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { + Phase(ctx, "build m") + close(entered) + <-release + return nil, c.runErr + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + <-entered + waitFor(t, "a step-progress in the build", func() bool { + pr := sy.of("step-progress") + return len(pr) > 0 && pr[len(pr)-1].body["phase"] == "build" + }) + close(release) + waitFor(t, "step-ended", func() bool { return len(sy.of("step-ended")) == 1 }) + taken := sy.of("ask-taken") + if len(taken) != 1 || taken[0].body["ask"] != a.ID || taken[0].body["node"] != "shanks" || taken[0].id != "ask-taken."+a.ID { + t.Fatalf("ask-taken = %+v, want one for %s on shanks", taken, a.ID) + } + if pr := sy.of("step-progress")[len(sy.of("step-progress"))-1].body; pr["ask"] != a.ID || pr["phase"] != "build" || pr["asker"] != a.Asker { + t.Fatalf("step-progress = %v, want the ask, its asker and the phase build", pr) + } + end := sy.of("step-ended")[0].body + if end["outcome"] != c.outcome || end["ask"] != a.ID || end["asker"] != a.Asker || end["node"] != "shanks" || end["at"] == nil { + t.Fatalf("step-ended = %v, want outcome %s of ask %s", end, c.outcome, a.ID) + } + if c.runErr != nil && end["said"] != "the build broke" { + t.Fatalf("step-ended said %v, want why", end["said"]) + } + }) + } +} diff --git a/modules/mesh-delivery-sdk/steprun/steprun.go b/modules/mesh-delivery-sdk/steprun/steprun.go index c8a44c0a..7cd29e6e 100644 --- a/modules/mesh-delivery-sdk/steprun/steprun.go +++ b/modules/mesh-delivery-sdk/steprun/steprun.go @@ -250,3 +250,15 @@ func SayStepOn(language, node string, publish func(subject string, body any, id return publish(subject, json.RawMessage(body), "step-ended."+end.ID) } } + +// SayOn is a holder's Puller.Say: an event of the language's build seat published on that seat's subject of node with +// publish (stdio.SeatPublish), `mesh.seat.node-build-.event..`. +func SayOn(language, node string, publish func(subject string, body any, id string) error) func(event string, body any, id string) error { + return func(event string, body any, id string) error { + seat, err := sdkstep.BuildSeat(language) + if err != nil { + return err + } + return publish("mesh.seat."+seat+".event."+event+"."+node, body, id) + } +}