diff --git a/modules/mesh-build-go/buildstep/pull.go b/modules/mesh-build-go/buildstep/pull.go index b811bb897..d9bec6b3e 100644 --- a/modules/mesh-build-go/buildstep/pull.go +++ b/modules/mesh-build-go/buildstep/pull.go @@ -703,8 +703,8 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key step = ph.end() // its end keeps the phase it ended in close(done) if d := droppedCause(runCtx); d != nil && ctx.Err() == nil { - // The seat's verb drop ended it (ADR 0664 §5): refused, naming who dropped it and why; acknowledged, never given again. - err = &Refusal{Text: d.Error()} + // The seat's verb drop ended it (ADR 0664 §4, §5): dropped, naming who dropped it and why; acknowledged, never given again. + err = &Dropped{Text: d.Error()} defer p.forgetDrop(ask.ID) } p.mu.Lock() diff --git a/modules/mesh-build-go/buildstep/pull_drop.go b/modules/mesh-build-go/buildstep/pull_drop.go index b664802e6..ffc4187da 100644 --- a/modules/mesh-build-go/buildstep/pull_drop.go +++ b/modules/mesh-build-go/buildstep/pull_drop.go @@ -8,6 +8,7 @@ import ( "strings" "time" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" buildrun "git.novox.be/novox/mesh-delivery/sdk/buildrun" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" stdio "git.novox.be/novox/mesh-sdk/go" @@ -16,7 +17,7 @@ import ( // **The seat's verb drop ends an ask its caller no longer waits for** (novox/hq ADR 0664 §5, issue 1748; research 086 // choice 28's cancel). It writes the ask in this node's own pauses state under dropped. (the part build-step), so a // holder that restarts still refuses the ask when the bus gives it again, and ends the step if it runs here: the step's -// context is cancelled, its process ended, and its key is written refused naming who dropped it and why. The state is the +// context is cancelled, its process ended, and its key is written dropped naming who dropped it and why. The state is the // module's, shared by every node: any holder answers the verb, every node refuses the ask when it pulls it (an ask that // waits leaves the stream at its next pull, with no work done), and the node that runs it hears the key by its watch // (watchDrops) and ends the step. @@ -62,7 +63,8 @@ func (p *Puller) Drop(id, by, why string) error { if strings.TrimSpace(id) == "" || strings.TrimSpace(why) == "" { return errors.New("buildstep: give the ask's id and why it is dropped") } - raw, err := json.Marshal(sdkstep.Drop{V: sdkstep.Version, By: by, Why: why, At: now().UTC()}) + at := now().UTC() + raw, err := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), By: by, Why: why, At: &at}) if err != nil { return err } @@ -95,7 +97,7 @@ func (p *Puller) watchDrops() { if !ok || ch.Op != "put" { return nil } - var d sdkstep.Drop + var d nodebuildgo.Drop if json.Unmarshal(ch.Value, &d) != nil { return nil } @@ -113,7 +115,7 @@ func (p *Puller) watchDrops() { } // droppedAsk is the drop kept for the ask, nil for none (or no state to read it from). -func (p *Puller) droppedAsk(id string) *sdkstep.Drop { +func (p *Puller) droppedAsk(id string) *nodebuildgo.Drop { if p.Load == nil { return nil } @@ -121,7 +123,7 @@ func (p *Puller) droppedAsk(id string) *sdkstep.Drop { if err != nil || raw == nil { return nil } - var d sdkstep.Drop + var d nodebuildgo.Drop if json.Unmarshal(raw, &d) != nil { return nil } @@ -130,11 +132,11 @@ func (p *Puller) droppedAsk(id string) *sdkstep.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 *sdkstep.Drop) { +func (p *Puller) refuseDropped(msg *Pulled, ask sdkstep.StepAsk, d *nodebuildgo.Drop) { 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.StepRefused, Started: at, Ended: at, Owner: buildrun.Owner, + step := sdkstep.Step{ID: ask.ID, Tool: ask.Kind + " ask", Node: p.Node, State: sdkstep.StepDropped, Started: at, Ended: at, Owner: buildrun.Owner, Run: int(msg.Delivered), Reason: (&droppedError{by: d.By, why: d.Why}).Error()} err = writeEnd(p.Put, key, step) } @@ -173,7 +175,7 @@ func (p *Puller) sweepDropped() { if !ok { continue } - if d := p.droppedAsk(id); d != nil && now().Sub(d.At) > DroppedKeep { + if d := p.droppedAsk(id); d != nil && (d.At == nil || now().Sub(*d.At) > DroppedKeep) { p.forgetDrop(id) } } diff --git a/modules/mesh-build-go/buildstep/pull_seat_test.go b/modules/mesh-build-go/buildstep/pull_seat_test.go index 6a59c264b..3afb7f71d 100644 --- a/modules/mesh-build-go/buildstep/pull_seat_test.go +++ b/modules/mesh-build-go/buildstep/pull_seat_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -110,7 +111,7 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) { if running { <-started // Another node's holder served the verb: its write reaches this node by the watch of the shared pauses state. - raw, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, By: "mesh-delivery-builder", Why: "superseded", At: time.Now()}) + raw, _ := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), By: "mesh-delivery-builder", Why: "superseded", At: ptr(time.Now())}) kv.put(sdkstep.DropKey(a.ID), json.RawMessage(raw)) if err := heard(stdio.StateChange{Key: sdkstep.DropKey(a.ID), Op: "put", Value: raw}); err != nil { t.Fatal(err) @@ -120,15 +121,15 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) { cancel() key, _ := sdkstep.AskStepKey("shanks", a.ID) s, ok := st.get(key) - if !ok || s.State != sdkstep.StepRefused || !strings.Contains(s.Reason, "dropped by mesh-delivery-builder: superseded") || ran != running { + 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) } waitFor(t, fmt.Sprint("the drop key deleted, running ", running), func() bool { kv.mu.Lock(); defer kv.mu.Unlock(); _, held := kv.m[sdkstep.DropKey(a.ID)]; return !held }) } // The start deletes a drop older than DroppedKeep and keeps a newer one. kv := &pauseKV{} - old, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, Why: "old", At: time.Now().Add(-DroppedKeep - time.Hour)}) - fresh, _ := json.Marshal(sdkstep.Drop{V: sdkstep.Version, Why: "fresh", At: time.Now()}) + old, _ := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), Why: "old", At: ptr(time.Now().Add(-DroppedKeep - time.Hour))}) + fresh, _ := json.Marshal(nodebuildgo.Drop{V: int64(sdkstep.Version), Why: "fresh", At: ptr(time.Now())}) kv.put(sdkstep.DropKey("old"), json.RawMessage(old)) kv.put(sdkstep.DropKey("fresh"), json.RawMessage(fresh)) p := &Puller{Load: kv.load, ForgetPause: kv.forget, PauseKeys: kv.keys} @@ -137,3 +138,5 @@ func TestADroppedAskIsRefusedAtPullAndARunningOneIsEnded(t *testing.T) { t.Errorf("keys after the sweep = %v, want the fresh drop alone", keys) } } + +func ptr[T any](v T) *T { return &v } diff --git a/modules/mesh-build-go/buildstep/refusal.go b/modules/mesh-build-go/buildstep/refusal.go index 909b2e6ea..d1f790dc1 100644 --- a/modules/mesh-build-go/buildstep/refusal.go +++ b/modules/mesh-build-go/buildstep/refusal.go @@ -27,3 +27,9 @@ type Declining struct { } func (d *Declining) Error() string { return d.Text } + +// Dropped is a step the seat's verb drop ended: its asker no longer waited for it. A step's run ends with one and the runner +// writes the step dropped with its text, the cause its caller named (ADR 0664 §4). +type Dropped struct{ Text string } + +func (d *Dropped) Error() string { return d.Text } diff --git a/modules/mesh-build-go/buildstep/step.go b/modules/mesh-build-go/buildstep/step.go index 5981000b3..47403fdfb 100644 --- a/modules/mesh-build-go/buildstep/step.go +++ b/modules/mesh-build-go/buildstep/step.go @@ -73,7 +73,7 @@ func end(ctx context.Context, put Put, key string, step sdkstep.Step, run func(c // the step's ask is not acknowledged as done. func finish(put Put, key string, step sdkstep.Step, result any, err error, logf func(format string, args ...any)) error { step.Ended = now().UTC() - switch ref := (*Refusal)(nil); { + switch ref, drop := (*Refusal)(nil), (*Dropped)(nil); { case err == nil: step.State = sdkstep.StepPassed if result != nil { @@ -84,6 +84,8 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf step.Result = raw } } + case errors.As(err, &drop): + step.State, step.Reason = sdkstep.StepDropped, err.Error() case errors.As(err, &ref): step.State, step.Reason = sdkstep.StepRefused, err.Error() default: