carlos / platform Public

host: the sealer's revert reports the durable lift before the cleanup, and the seal records its final copy

offlineSealRevert writes the reverting report (TombstoneLifted, the first
lift time kept on every retry) once lease.LiftTombstone has answered and
before the target's prefixes are deleted; the final reverted report keeps
both lift fields. The seal records FinalCopiedBytes (the final round's
copied bytes) and the box's identity; the sealer's other reports go
through putOfflineReport.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

claude 2 days ago ea8fc80e536ccfa12c2596bcfba7481431caeb28 parent ad8de81

1 file changed, +41 −7

internal/host/movesoffline_seal.go +41 −7

diff --git a/internal/host/movesoffline_seal.go b/internal/host/movesoffline_seal.go

index f72f38d61..d269cd60d 100644

--- a/internal/host/movesoffline_seal.go

+++ b/internal/host/movesoffline_seal.go

@@ -200,11 +200,22 @@ func (w *movesWorker) offlineLock(ctx context.Context, host string) (tombState,

// offlineTarget is the target's coordinates as its edges file publishes

// them: the box at move.to (the lexically first, as every move reads the

// fleet), matched exactly and then by placement under the home binding.

+//

+// A move with pinned participants (a self-serve move) is the target

+// participant's box and no other: its edges file by box id, whatever else

+// the target region publishes.

func (w *movesWorker) offlineTarget(ctx context.Context, s release.Store, m *provision.MoveState) (edgesFile, error) {

fleet, err := loadMoveFleet(ctx, s)

if err != nil {

return edgesFile{}, fmt.Errorf("read the fleet's edges files: %w", err)

}

+ if m.Participants != nil {

+ p := m.Participants.Target

+ if t, ok := fleet.byBox[p.BoxID]; ok && p.BoxID != "" {

+ return t, nil

+ }

+ return edgesFile{}, fmt.Errorf("the target box %s publishes no data bucket yet", p.Label)

+ }

if t, ok := fleet.targets[m.To]; ok {

return t, nil

}

@@ -279,7 +290,7 @@ func (w *movesWorker) offlineSeal(ctx context.Context, s release.Store, rec prov

items[ref.item] = moves.Check{OK: false, Reason: ref.Reason}

r := moves.OfflineReport{Op: m.ID, Host: host, Box: w.cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseFailed,

At: w.cfg.now().UTC(), Items: items}

- return errors.Join(fmt.Errorf("offline move %s: %s cannot be sealed: %w", m.ID, host, err), writeOfflineReport(ctx, s, r))

+ return errors.Join(fmt.Errorf("offline move %s: %s cannot be sealed: %w", m.ID, host, err), w.putOfflineReport(ctx, s, r))

}

if err != nil {

return fmt.Errorf("offline move %s: sealing %s: %w", m.ID, host, err)

@@ -606,6 +617,7 @@ func (w *movesWorker) offlineTakeSeal(ctx context.Context, s release.Store, rec

FilesPlain: a.plain, FilesSealed: a.sealed, FilesBytes: a.bytes, FilesEntries: a.entries,

EnvSHA256: envSHA, KeySHA256: keySHA, EnvDigest: envReport.Digest,

Release: version, TimersDigest: timersDigest, Replication: rt.Replication, NextDue: rt.NextDue,

+ FinalCopiedBytes: round.BytesCopied, BoxID: cfg.BoxID, InstanceID: cfg.InstanceID,

}

if err := moves.ValidateOfflineSeal(seal, host, m.ID); err != nil {

return seal, err

@@ -970,7 +982,7 @@ func (w *movesWorker) offlineKeepStop(ctx context.Context, s release.Store, rec

living.At = cfg.now().UTC()

living.TombstoneOwner = st.owner

living.TombstoneUntil = st.expires.UTC()

- return writeOfflineReport(ctx, s, living)

+ return w.putOfflineReport(ctx, s, living)

}

// offlineSealerRefresh is the sealer's half of the report refresh

@@ -1021,7 +1033,7 @@ func (w *movesWorker) offlineSealerRefresh(ctx context.Context, s release.Store,

living.At = cfg.now().UTC()

living.TombstoneOwner = st.owner

living.TombstoneUntil = st.expires.UTC()

- return writeOfflineReport(ctx, s, living)

+ return w.putOfflineReport(ctx, s, living)

}

// offlineCheckPositions is the backstop behind the holds and the fence:

@@ -1087,7 +1099,7 @@ func (w *movesWorker) offlineLostStop(ctx context.Context, s release.Store, rec

}

items[item] = moves.Check{OK: false, Reason: "lost stop: " + cause}

r.Items = items

- werr := writeOfflineReport(ctx, s, r)

+ werr := w.putOfflineReport(ctx, s, r)

subject, body := offlineLostStopPage(host, m.ID, m.State, cause)

w.firePage(ctx, host, m.ID, offlinePageLostStop, alerting.LevelIncident, subject, body)

log.Printf("moves: %s: offline move %s: LOST STOP: %s", host, m.ID, cause)

@@ -1164,7 +1176,7 @@ func (w *movesWorker) offlineFinishLift(ctx context.Context, s release.Store, re

}

clearOfflineReceipts(ctx, own, cfg.OwnBucket, host)

r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseLifted, At: cfg.now().UTC()}

- return writeOfflineReport(ctx, s, r)

+ return w.putOfflineReport(ctx, s, r)

}

// clearOfflineReceipts is clearMoveReceipts as an ensure: receipts that are

@@ -1192,6 +1204,16 @@ func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, re

if err != nil {

return err

}

+ // The lift as this box's reports have recorded it: the first time a

+ // lift was reported is kept on every retry, and on the final report.

+ prev, err := w.ownReport(ctx, s, rec, moves.OfflineRoleSealer)

+ if err != nil {

+ return fmt.Errorf("offline move %s: read the sealer report: %w", m.ID, err)

+ }

+ var liftedAt time.Time

+ if prev != nil && prev.Op == m.ID && prev.TombstoneLifted && !prev.TombstoneLiftedAt.IsZero() {

+ liftedAt = prev.TombstoneLiftedAt

+ }

lifted, err := lease.LiftTombstone(ctx, bucket, loc, region, func(owner string) bool { return owner == moves.TombstoneOwner(m.ID) })

var held *lease.HeldError

switch {

@@ -1199,6 +1221,17 @@ func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, re

if lifted {

log.Printf("moves: %s: offline move %s: tombstone lifted on abort", host, m.ID)

}

+ // The source's lease holds no tombstone of this move: say so,

+ // durably, before the cleanup of the target's prefixes, which may

+ // take long or fail and be retried.

+ if liftedAt.IsZero() {

+ liftedAt = cfg.now().UTC()

+ }

+ r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverting, At: cfg.now().UTC(),

+ BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, TombstoneLifted: true, TombstoneLiftedAt: liftedAt}

+ if err := w.putOfflineReport(ctx, s, r); err != nil {

+ return fmt.Errorf("offline move %s: report the lift of %s: %w", m.ID, host, err)

+ }

case errors.As(err, &held):

w.notice(host+"#offline-revert-lease", "moves: %s: offline move %s: the lease is %s's, not this move's; left as it is", host, m.ID, held.Owner)

default:

@@ -1225,6 +1258,7 @@ func (w *movesWorker) offlineSealRevert(ctx context.Context, s release.Store, re

}

clearOfflineReceipts(ctx, mirror, bucket, host)

w.offlineLiftFence(host, m)

- r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()}

- return writeOfflineReport(ctx, s, r)

+ r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleSealer, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC(),

+ BoxID: cfg.BoxID, InstanceID: cfg.InstanceID, TombstoneLifted: !liftedAt.IsZero(), TombstoneLiftedAt: liftedAt}

+ return w.putOfflineReport(ctx, s, r)

}