diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 3e2f748de..78987366b 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -179,8 +179,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro for { select { case <-ctx.Done(): + // A move or a restart stops the process: it takes no new call, the running act ends through its own failure + // path, and every accepted act ends stopped, so its caller asks again (§562, A-141). + e.Stop("its stop signal") stopAnswers() - e.Wait() return nil case <-tick.C: expire("the daily removal of points past their keep days (design 47 G section 551)") diff --git a/modules/mesh-recovery/internal/recovery/control.go b/modules/mesh-recovery/internal/recovery/control.go index ef9ad13ea..d4841a96c 100644 --- a/modules/mesh-recovery/internal/recovery/control.go +++ b/modules/mesh-recovery/internal/recovery/control.go @@ -17,6 +17,9 @@ import ( // refuses an act that has begun, naming its step, since a restore cut off after its removal is put back only by its own // failure path (§516). It is written in the hand-act log first and answers in §691's one form. func (e *Engine) Cancel(act, why, caller, asker string) Answer { + if stop := e.Stopping(); stop != "" { + return Answer{Error: stop} + } if err := e.Seats.HandAct(caller, asker, "cancel "+act, why); err != nil { return Answer{Error: "the hand-act log was not written, so nothing was done: " + err.Error()} } @@ -150,6 +153,13 @@ func (e *Engine) Start(ctx context.Context) []UnrestoredAside { } } } + e.endUnended() + e.judge() + return e.Unrestored() +} + +// endUnended ends every accepted act that never ended failed with `stopped` and its ending event (§554, §562), oldest first. +func (e *Engine) endUnended() { acts := e.readActs() sort.Slice(acts, func(i, j int) bool { return acts[i].Since.Before(acts[j].Since) }) for i := range acts { @@ -167,8 +177,6 @@ func (e *Engine) Start(ctx context.Context) []UnrestoredAside { a.Error = ErrStopped e.markEnded(&a) } - e.judge() - return e.Unrestored() } func anyExists(os []Object) bool { diff --git a/modules/mesh-recovery/internal/recovery/points.go b/modules/mesh-recovery/internal/recovery/points.go index 18e5ec14c..b3d9f1bed 100644 --- a/modules/mesh-recovery/internal/recovery/points.go +++ b/modules/mesh-recovery/internal/recovery/points.go @@ -52,6 +52,7 @@ func (e *Engine) say(event string, body any, parts ...string) { } func (e *Engine) endedFailed(a *Act, why error) error { + why = e.stoppedOr(why) e.say(seat.EventEnded, seat.EndedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, Fingerprint: a.Fingerprint, Snapshot: a.Snapshot, Outcome: seat.EndedEventOutcomeFailed, Error: why.Error(), Why: cancelWhy(a, why)}, a.Module, a.Kept) diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index 2e1f09076..efd0faddf 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -131,6 +131,7 @@ type Engine struct { wg sync.WaitGroup seq int why string // the hand act words of a cancel (A-117) + stop string // why the process stops, once it does (§562, A-141) } // Act is one accepted act, written in the directory before its caller is answered (§554). @@ -242,6 +243,9 @@ var errBusy = errors.New("busy") func (e *Engine) Accept(a Act) Answer { e.mu.Lock() defer e.mu.Unlock() + if e.stop != "" { + return Answer{Error: e.stop} + } if e.running != nil { return Answer{Busy: e.running.ID} } @@ -275,6 +279,36 @@ func (e *Engine) Accept(a Act) Answer { // Wait returns once the running act, if any, has ended (a whole test's and a stop's join). func (e *Engine) Wait() { e.wg.Wait() } +// Stop is the process's stop, a move's or a restart's (§562, A-141): from now on every call is answered an error naming the +// stop, the running act's context is cancelled so it ends through its own failure path (a restore past its removal puts +// the aside back and sets the consumers back), and once it has ended every accepted act not ended is ended failed with +// `stopped` and its ending event, so its caller asks again. +func (e *Engine) Stop(why string) { + e.mu.Lock() + e.stop = "the recovery module is stopping (" + why + "), so it takes no call; ask again once it runs" + if e.cancel != nil { + e.cancel() + } + e.mu.Unlock() + e.wg.Wait() + e.endUnended() +} + +// Stopping is the error a call is answered while the process stops, empty while it runs. +func (e *Engine) Stopping() string { + e.mu.Lock() + defer e.mu.Unlock() + return e.stop +} + +// stoppedOr is an act's failure as its ending event says it: `stopped` once the process stops (§562). +func (e *Engine) stoppedOr(err error) error { + if err != nil && e.Stopping() != "" { + return errors.New(ErrStopped) + } + return err +} + func (e *Engine) setStep(a *Act, step, waitsOn string) { e.mu.Lock() a.Step, a.WaitsOn = step, waitsOn diff --git a/modules/mesh-recovery/internal/recovery/restore.go b/modules/mesh-recovery/internal/recovery/restore.go index c6936f758..6fd4fd60e 100644 --- a/modules/mesh-recovery/internal/recovery/restore.go +++ b/modules/mesh-recovery/internal/recovery/restore.go @@ -54,6 +54,7 @@ func (e *Engine) asideRoot(module, kept string) (string, error) { func (e *Engine) endedRestored(a *Act, outcomeErr error, aside *seat.Aside, missing bool) error { ev := seat.RestoredEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, Fingerprint: a.Fingerprint, Snapshot: a.Snapshot, Outcome: seat.RestoredEventOutcomeDone, Aside: aside, Missing: missing} + outcomeErr = e.stoppedOr(outcomeErr) if outcomeErr != nil { ev.Outcome, ev.Error, ev.Why = seat.RestoredEventOutcomeFailed, outcomeErr.Error(), e.cancelledWhy(outcomeErr) } @@ -257,8 +258,10 @@ func (e *Engine) putBack(ctx context.Context, dir string, ds []seat.Digest, cs [ return e.Bus.SetConsumers(ctx, cs) } -// restoreFromAside puts an aside back under the live names: any partly restored stream is removed first. +// restoreFromAside puts an aside back under the live names: any partly restored stream is removed first. It is a failure +// path, so it runs whole also when the act's context ended, as at the process's stop (§562). func (e *Engine) restoreFromAside(ctx context.Context, rec *asideRecord, clear bool) error { + ctx = context.WithoutCancel(ctx) if clear { for _, s := range rec.Streams { if err := e.Bus.Delete(ctx, s); err != nil { diff --git a/modules/mesh-recovery/internal/recovery/restore_test.go b/modules/mesh-recovery/internal/recovery/restore_test.go index 88031c462..cc168ee6c 100644 --- a/modules/mesh-recovery/internal/recovery/restore_test.go +++ b/modules/mesh-recovery/internal/recovery/restore_test.go @@ -294,3 +294,52 @@ func TestARestoreWhoseStartIsNotAcknowledgedTouchesNothing(t *testing.T) { t.Errorf("an aside was taken: %+v", kp.Asides) } } + +// stallBus blocks its first Restore until the act's context ends, as a restore under way when the process stops. +type stallBus struct { + *fakeBus + in chan struct{} + stalled bool +} + +func (s *stallBus) Restore(ctx context.Context, file string) error { + if !s.stalled { + s.stalled = true + close(s.in) + <-ctx.Done() + return ctx.Err() + } + return s.fakeBus.Restore(ctx, file) +} + +// A stop of the process (a move's or a restart's) answers every new call an error naming it, ends the running restore +// through its failure path, putting the aside back, and ends every accepted act with `stopped`, so its caller asks again +// (design 47 G §562, A-141). +func TestAStopEndsTheRunningActThroughItsFailurePathAndStopped(t *testing.T) { + r := takenRig(t) + r.b.streams["app_notes"] = "changed since" + stall := &stallBus{fakeBus: r.b, in: make(chan struct{})} + r.e.Bus = stall + ans := r.e.Accept(restoreAct("app/notes/2-fp1")) + <-stall.in + // An accepted act a crash left unended is ended by the stop too. + _ = os.WriteFile(filepath.Join(r.e.Dir, "acts", "act-0.json"), []byte(`{"act":"act-0","kind":"take","module":"app","kept":"seen","why":"w","caller":"c","since":"2026-11-11T00:00:00Z"}`), 0o600) + r.e.Stop("a move") + if r.b.streams["app_notes"] != "changed since" || r.b.streams["KV_app_seen"] != "world" { + t.Errorf("the aside is not put back: %v", r.b.streams) + } + if ev := lastRestored(t, r); ev.Act != ans.Accepted || ev.Outcome != "failed" || ev.Error != recovery.ErrStopped { + t.Errorf("restored %+v", ev) + } + ended := r.s.named(seat.EventEnded) + if len(ended) != 1 || ended[0].Body.(seat.EndedEvent).Act != "act-0" || ended[0].Body.(seat.EndedEvent).Error != recovery.ErrStopped { + t.Errorf("ended %+v", ended) + } + if got := r.e.Accept(takeAct()); !strings.Contains(got.Error, "stopping (a move)") { + t.Errorf("a call during the stop answered %+v", got) + } + verb := r.e.Handlers()[seat.VerbKeptPoints] + if got, _ := verb("c", []byte(`{}`)); !strings.Contains(got.(recovery.Answer).Error, "stopping") { + t.Errorf("kept-points during the stop answered %+v", got) + } +} diff --git a/modules/mesh-recovery/internal/recovery/verbs.go b/modules/mesh-recovery/internal/recovery/verbs.go index 34cbdf7ec..bd053a9ca 100644 --- a/modules/mesh-recovery/internal/recovery/verbs.go +++ b/modules/mesh-recovery/internal/recovery/verbs.go @@ -77,6 +77,15 @@ func (e *Engine) Handlers() map[string]process.SeatHandler { } return e.Cancel(in.Act, in.Why, caller, in.Asker), nil } + // A stopping process takes no call, each answered the error naming the stop (§562, A-141). + for verb, handle := range h { + h[verb] = func(caller string, body json.RawMessage) (any, error) { + if stop := e.Stopping(); stop != "" { + return Answer{Error: stop}, nil + } + return handle(caller, body) + } + } return h }