pool: a live move aborts a paused final round that overruns max_downtime_ms, and reports its phases

docs/cluster.md

@@ -450,9 +450,22 @@ the rounds reached, plus 400 ms for the receive and the start, fits
 never fit), the source pauses the guest for the final
 round and the target receives the series and starts the guest. A move whose
 deltas never fit answers 503 and the guest runs on; from the second delta on, one
-that shrinks by under 10% and is still too large answers at once. `downtime_ms` in the answer
+that shrinks by under 10% and is still too large answers at once. The estimate
+cannot see the final capture, which runs with the guest paused, or the pages
+dirtied since the last delta, so once the guest is paused the source checks again: the final
+capture's time plus the final round's bytes on the wire (the last delta's, plus
+the growth of the final round) at the rounds' rate plus the 400 ms. Past `max_downtime_ms` it aborts the
+move before sending the final round: `cocoon vm migrate abort` resumes the guest
+on the source, the target drops what it received, the record ends `aborted`, and
+the move answers 503. The check cannot give back a pause already spent, so a final
+capture that alone overruns the budget still pauses the guest for the capture plus
+the abort and its record write; with the meta store up, that, not
+`max_downtime_ms`, is the worst case. Past the check, a slow
+transfer or commit still lengthens the pause, and `downtime_ms` reports it. `downtime_ms` in the answer
 is what the source measured from the final capture's start, including its wait
-for a free dump slot, until the target answers the commit, after it started the guest. Each round has one minute plus its size at 10 MB/s to arrive, so a
+for a free dump slot, until the target answers the commit, after it started the guest;
+of which `capture_ms`, `send_ms` and `commit_ms` are the final capture, the final
+round's transfer and the commit. Each round has one minute plus its size at 10 MB/s to arrive, so a
 target that stops reading fails the move. A replicated claim's registry unrow, the final capture, its transfer and the
 target's commit have 2 minutes together; past that the source resolves the move through the record, and the
 guest resumes on the source unless the target committed.

docs/sandboxd-api.md

@@ -914,10 +914,10 @@ its guest's processes and memory (see
 `{"to": "<peer>", "max_downtime_ms": 2000}`, `to` as `GET /v1/peers` lists it
 and never this node (else 400), `max_downtime_ms` optional with a default of
 2000 and a floor: 400 or less answers 400 at once, because the target's receive
-and start alone are allowed that long. Answers `{"owner_addr": "<peer>", "rounds": n, "downtime_ms": d}` once the
+and start alone are allowed that long. Answers `{"owner_addr": "<peer>", "rounds": n, "downtime_ms": d, "capture_ms": c, "send_ms": s, "commit_ms": m}` once the
 peer took the claim and this node dropped it; a guest the peer cannot start is
 failed there. `downtime_ms` runs from the final capture's start, including its
-wait for a free dump slot, until the peer answers the commit, after it started the guest. While the move runs, release, fork,
+wait for a free dump slot, until the peer answers the commit, after it started the guest; of it, `capture_ms`, `send_ms` and `commit_ms` are the final capture, the final round transfer and the commit. A final round whose capture time and size no longer fit `max_downtime_ms` is aborted before it is sent and the guest resumes here, so with the meta store up a capture that alone overruns pauses the guest for the capture plus the abort and its record write. While the move runs, release, fork,
 checkpoint, promote, hibernate and volume changes of the claim answer 503
 `sandbox moving to another node; retry`. Env writes, renew and data-plane calls keep working until
 the final round and answer the same 503 during it, because the peer would miss
@@ -938,7 +938,8 @@ node drops the claim (see [moving a running claim](cluster.md#moving-a-running-c
 - 502 the target is unreachable or failed
 - 503 the move or archive registry is down, the node lease is overdue, the node
   is evacuating the claim (`sandbox moving to another node; retry`), or the
-  final round would not fit `max_downtime_ms`, after at most 4 deltas
+  final round would not fit `max_downtime_ms`, after at most 4 deltas or once
+  its capture overran it with the guest paused (the guest resumes here)
 
 The target serves root-only peer routes for a move, `POST /v1/moves/{id}`,
 `PUT /v1/moves/{id}/rounds/{n}`, `POST /v1/moves/{id}/commit` and

sandboxd/pool/move.go

@@ -38,6 +38,9 @@ type MoveAdmission struct {
 type MoveResult struct {
 	Rounds   int
 	Downtime time.Duration
+	Capture  time.Duration
+	Send     time.Duration
+	Commit   time.Duration
 }
 
 // MovePeer reaches the target node of a live move.
@@ -58,14 +61,17 @@ type outMove struct {
 	final    bool
 	locked   bool
 	pausedAt time.Time
+	capture  time.Duration
+	send     time.Duration
+	commit   time.Duration
 }
 
 func (mv *outMove) owns(rec moves.Record, node string) bool {
 	return rec.From == node && rec.Epoch >= mv.Epoch
 }
 
 func (mv *outMove) result() MoveResult {
-	res := MoveResult{Rounds: mv.rounds}
+	res := MoveResult{Rounds: mv.rounds, Capture: mv.capture, Send: mv.send, Commit: mv.commit}
 	if !mv.pausedAt.IsZero() {
 		res.Downtime = time.Since(mv.pausedAt)
 	}

sandboxd/pool/move_out.go

@@ -74,13 +74,12 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b
 	if err = m.journalMoveOut(ctx, sb, mv); err != nil {
 		return err
 	}
-	var sent, previous int64
+	var size, sent, previous int64
 	var sending time.Duration
 	for delta := 0; ; delta++ {
 		start := time.Now()
-		size, sendErr := m.sendRound(ctx, sb.ID, mv, round)
-		if sendErr != nil {
-			return sendErr
+		if size, err = m.sendRound(ctx, sb.ID, mv, round); err != nil {
+			return err
 		}
 		sending += time.Since(start)
 		sent += size
@@ -97,6 +96,7 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b
 			return fmt.Errorf("%w: delta capture: %w", ErrNotMovable, err)
 		}
 	}
+	lastBytes := roundBytes(round)
 	m.mu.Lock()
 	mv.final = true
 	m.mu.Unlock()
@@ -111,13 +111,21 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b
 	if round, err = m.eng.MigrateCapture(ctx, vm, mv.Series, true); err != nil {
 		return fmt.Errorf("%w: final capture: %w", ErrNotMovable, err)
 	}
+	mv.capture = time.Since(mv.pausedAt)
+	apparent := roundBytes(round)
+	wire := size + max(apparent-lastBytes, 0)
+	if pause := mv.capture + finalPause(wire, sent, sending); pause > budget {
+		return fmt.Errorf("%w: the final round of %d bytes took %s to capture, leaving a pause of about %s", ErrMoveDowntime, apparent, mv.capture.Round(time.Millisecond), pause.Round(time.Millisecond))
+	}
 	final, err := m.moves.Advance(ctx, sb.ID, mv.Epoch, moves.Pending, moves.Final)
 	if err != nil {
 		return err
 	}
+	start := time.Now()
 	if _, err = m.sendRound(ctx, sb.ID, mv, round); err != nil {
 		return err
 	}
+	mv.send = time.Since(start)
 	if used, ok := m.out.TunUsed(sb.ID); ok {
 		m.setTunBytes(sb, used)
 	}
@@ -128,7 +136,10 @@ func (m *Manager) moveOut(ctx context.Context, sb *types.Sandbox, mv *outMove, b
 	if err != nil {
 		return fmt.Errorf("encode claim: %w", err)
 	}
-	return m.movePeer.Commit(ctx, mv.Addr, sb.ID, claim, final)
+	start = time.Now()
+	err = m.movePeer.Commit(ctx, mv.Addr, sb.ID, claim, final)
+	mv.commit = time.Since(start)
+	return err
 }
 
 func (m *Manager) beginMove(ctx context.Context, sb *types.Sandbox, to string) (int64, error) {
@@ -155,11 +166,10 @@ func (m *Manager) beginMove(ctx context.Context, sb *types.Sandbox, to string) (
 func (m *Manager) sendRound(ctx context.Context, id string, mv *outMove, round types.MigrateRound) (int64, error) {
 	mv.rounds++
 	names := make([]string, len(round.Files))
-	var size int64
 	for i, f := range round.Files {
-		names[i], size = f.Name, size+f.Size
+		names[i] = f.Name
 	}
-	ctx, cancel := context.WithTimeout(ctx, moveRoundBase+time.Duration(size/moveRoundMinRate)*time.Second)
+	ctx, cancel := context.WithTimeout(ctx, moveRoundBase+time.Duration(roundBytes(round)/moveRoundMinRate)*time.Second)
 	defer cancel()
 	return m.movePeer.SendRound(ctx, mv.Addr, id, mv.rounds, round.Dir, names)
 }
@@ -413,6 +423,14 @@ func (m *Manager) midMoveLocked(sb *types.Sandbox) bool {
 	return moving || sb.MoveIn
 }
 
+func roundBytes(round types.MigrateRound) int64 {
+	var size int64
+	for _, f := range round.Files {
+		size += f.Size
+	}
+	return size
+}
+
 func finalPause(last, sent int64, sending time.Duration) time.Duration {
 	return MoveStartAllowance + time.Duration(float64(sending)*float64(last)/float64(sent))
 }

sandboxd/pool/move_test.go

@@ -43,8 +43,8 @@ func TestMoveHandsARunningClaimToThePeerWithBothNodesCountingIt(t *testing.T) {
 	if err != nil {
 		t.Fatalf("move: %v", err)
 	}
-	if res.Rounds != 3 || res.Downtime <= 0 {
-		t.Errorf("result %+v, want a baseline, one delta and the final round with a downtime", res)
+	if res.Rounds != 3 || res.Downtime <= 0 || res.Capture <= 0 || res.Send <= 0 || res.Commit <= 0 || res.Capture+res.Send+res.Commit > res.Downtime {
+		t.Errorf("result %+v, want a baseline, one delta and the final round with a downtime that covers its phases", res)
 	}
 	for i, counts := range during {
 		if counts != [2][3]int{{1, 1, 1}, {1, 1, 1}} {
@@ -741,6 +741,7 @@ func TestMoveGivesEachRoundADeadlineFromItsSize(t *testing.T) {
 	mc := newMoveCluster(t)
 	sb := mustClaim(t, mc.s, testKey)
 	fakeOf(mc.s).migrateSizes = []int64{1 << 30, 1, 1 << 40}
+	mc.p.wireSizes = []int64{1 << 30, 0}
 	if _, err := mc.s.Move(t.Context(), sb.ID, "b", time.Hour); err != nil {
 		t.Fatalf("move: %v", err)
 	}
@@ -883,15 +884,78 @@ func TestMoveEstimatesThePauseFromTheBytesOnTheWire(t *testing.T) {
 	if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+50*time.Millisecond); !errors.Is(err, ErrMoveDowntime) {
 		t.Fatalf("move of a hot guest: %v, want ErrMoveDowntime", err)
 	}
-	eng.mu.Lock()
-	wentFinal := slices.ContainsFunc(eng.captures, func(c string) bool { return strings.HasSuffix(c, ":true") })
-	eng.mu.Unlock()
-	if wentFinal {
+	if finalCaptured(mc.s) {
 		t.Error("the move paused the guest for a final round that cannot fit")
 	}
 	assertStillOnSource(t, mc, sb)
 }
 
+func TestMoveWhosePausedRoundOverrunsTheBudgetResumesTheGuestOnTheSource(t *testing.T) {
+	for name, tt := range map[string]struct {
+		captureTime time.Duration
+		sizes, wire []int64
+	}{
+		"final capture":     {captureTime: 200 * time.Millisecond},
+		"final round grown": {sizes: []int64{1 << 20, 1 << 10, 1 << 40}, wire: []int64{1 << 20, 1 << 10}},
+	} {
+		t.Run(name, func(t *testing.T) {
+			mc := newMoveCluster(t)
+			sb := mustClaim(t, mc.s, testKey)
+			eng := fakeOf(mc.s)
+			eng.finalCaptureTime, eng.migrateSizes, mc.p.wireSizes = tt.captureTime, tt.sizes, tt.wire
+			sentFinal, committed := false, false
+			mc.p.roundHook = func(round int) error {
+				sentFinal = sentFinal || round == 3
+				return nil
+			}
+			mc.p.commitHook = func(int) error { committed = true; return nil }
+			if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+100*time.Millisecond); !errors.Is(err, ErrMoveDowntime) {
+				t.Fatalf("move: %v, want ErrMoveDowntime", err)
+			}
+			if !finalCaptured(mc.s) || sentFinal || committed {
+				t.Errorf("final capture %v sent %v commit %v, want the capture and neither the final round nor a commit", finalCaptured(mc.s), sentFinal, committed)
+			}
+			assertStillOnSource(t, mc, sb)
+			assertNoMoveResidue(t, mc.t, sb.ID)
+			if _, ok, _ := mc.moveReg.Get(t.Context(), sb.ID); ok {
+				t.Error("the aborted move left its record")
+			}
+		})
+	}
+}
+
+func TestMoveWhoseFinalRoundOutgrowsATinyDeltaStillFits(t *testing.T) {
+	mc := newMoveCluster(t)
+	sb := mustClaim(t, mc.s, testKey)
+	fakeOf(mc.s).migrateSizes = []int64{1 << 20, 1, 1 << 24}
+	mc.p.wireSizes = []int64{1 << 20, 1 << 12}
+	if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+500*time.Millisecond); err != nil {
+		t.Fatalf("move whose final round grew by 16 MiB over a 1-byte delta sent as 4 KiB: %v", err)
+	}
+}
+
+func TestMoveWhoseFinalRoundShrinksStillCountsTheLastDeltasBytes(t *testing.T) {
+	mc := newMoveCluster(t)
+	sb := mustClaim(t, mc.s, testKey)
+	eng := fakeOf(mc.s)
+	eng.finalCaptureTime, eng.migrateSizes = 100*time.Millisecond, []int64{1 << 30, 512 << 20, 256 << 20}
+	mc.p.wireSizes = []int64{32 << 20, 32 << 20}
+	mc.p.roundHook = func(round int) error {
+		if round == 2 {
+			time.Sleep(200 * time.Millisecond)
+		}
+		return nil
+	}
+	if _, err := mc.s.Move(t.Context(), sb.ID, "b", MoveStartAllowance+160*time.Millisecond); !errors.Is(err, ErrMoveDowntime) {
+		t.Fatalf("move whose sparse final round shrank after a slow 32 MiB delta: %v, want ErrMoveDowntime", err)
+	}
+	if !finalCaptured(mc.s) {
+		t.Error("the move refused before the final capture, want the paused-round check to refuse it")
+	}
+	assertStillOnSource(t, mc, sb)
+	assertNoMoveResidue(t, mc.t, sb.ID)
+}
+
 func TestMoveGivesUpOnDeltasThatStopShrinking(t *testing.T) {
 	for name, tt := range map[string]struct {
 		sizes    []int64
@@ -1874,6 +1938,13 @@ func movingOn(m *Manager, sb *types.Sandbox) bool {
 	return m.midMoveLocked(sb)
 }
 
+func finalCaptured(m *Manager) bool {
+	eng := fakeOf(m)
+	eng.mu.Lock()
+	defer eng.mu.Unlock()
+	return slices.ContainsFunc(eng.captures, func(c string) bool { return strings.HasSuffix(c, ":true") })
+}
+
 func pausedOn(m *Manager, vm string) bool {
 	eng := fakeOf(m)
 	eng.mu.Lock()

sandboxd/pool/pool_test.go

@@ -1481,6 +1481,7 @@ type fakeEngine struct {
 	migrateSupportErr error
 	captureErr        error
 	finalCaptureErr   error
+	finalCaptureTime  time.Duration
 	receiveErr        error
 	receiveStall      chan struct{}
 	receiveEntered    chan struct{}
@@ -2056,6 +2057,9 @@ func (f *fakeEngine) Balloon(_ context.Context, vmName, target string) (types.VM
 }
 
 func (f *fakeEngine) MigrateCapture(_ context.Context, vmName, series string, final bool) (types.MigrateRound, error) {
+	if final {
+		time.Sleep(f.finalCaptureTime)
+	}
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.captures = append(f.captures, fmt.Sprintf("%s:%s:%v", vmName, series, final))

sandboxd/server/move.go

@@ -111,6 +111,9 @@ type MoveResponse struct {
 	OwnerAddr  string `json:"owner_addr"`
 	Rounds     int    `json:"rounds"`
 	DowntimeMS int64  `json:"downtime_ms"`
+	CaptureMS  int64  `json:"capture_ms"`
+	SendMS     int64  `json:"send_ms"`
+	CommitMS   int64  `json:"commit_ms"`
 }
 
 func (s *Server) handleMove(w http.ResponseWriter, r *http.Request) {
@@ -131,7 +134,10 @@ func (s *Server) handleMove(w http.ResponseWriter, r *http.Request) {
 	id := r.PathValue("id")
 	res, err := s.mgr.Move(r.Context(), id, addr, budget)
 	writeResult(w, r, "move", id, "move failed", err, func() {
-		writeJSON(w, http.StatusOK, MoveResponse{OwnerAddr: s.placer.ClientAddr(addr), Rounds: res.Rounds, DowntimeMS: res.Downtime.Milliseconds()})
+		writeJSON(w, http.StatusOK, MoveResponse{
+			OwnerAddr: s.placer.ClientAddr(addr), Rounds: res.Rounds, DowntimeMS: res.Downtime.Milliseconds(),
+			CaptureMS: res.Capture.Milliseconds(), SendMS: res.Send.Milliseconds(), CommitMS: res.Commit.Milliseconds(),
+		})
 	})
 }
 

sandboxd/server/move_test.go

@@ -20,7 +20,7 @@ func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) {
 	var outcome error
 	mgr := &fakeManager{move: func(_, addr string, budget time.Duration) (pool.MoveResult, error) {
 		gotAddr, gotBudget = addr, budget
-		return pool.MoveResult{Rounds: 3, Downtime: 180 * time.Millisecond}, outcome
+		return pool.MoveResult{Rounds: 3, Downtime: 180 * time.Millisecond, Capture: 60 * time.Millisecond, Send: 90 * time.Millisecond, Commit: 25 * time.Millisecond}, outcome
 	}}
 	srv := newMeshTestServer(t, "root", mgr, &fakePlacer{addrs: []string{"10.0.0.2:7777"}}, nil)
 
@@ -30,7 +30,7 @@ func TestMoveRouteResolvesThePeerAndMapsTheOutcome(t *testing.T) {
 	if err := json.UnmarshalRead(resp.Body, &body); err != nil || resp.StatusCode != http.StatusOK {
 		t.Fatalf("move: %d %v", resp.StatusCode, err)
 	}
-	if gotAddr != "10.0.0.2:7777" || gotBudget != 1500*time.Millisecond || body != (MoveResponse{OwnerAddr: "10.0.0.2:7777", Rounds: 3, DowntimeMS: 180}) {
+	if gotAddr != "10.0.0.2:7777" || gotBudget != 1500*time.Millisecond || body != (MoveResponse{OwnerAddr: "10.0.0.2:7777", Rounds: 3, DowntimeMS: 180, CaptureMS: 60, SendMS: 90, CommitMS: 25}) {
 		t.Errorf("move to %s within %s answered %+v", gotAddr, gotBudget, body)
 	}
 	for _, tt := range []struct {