| index 534eb1d00..00079e33c 100644 |
| --- a/internal/host/movesphase1.go |
| +++ b/internal/host/movesphase1.go |
| @@ -121,6 +121,11 @@ type moveLedger struct { |
| // copied at least crossMinBytes, or the recopy that refreshed it. |
| LastRoundSeconds float64 `json:"last_round_seconds,omitempty"` |
| LastRoundAt time.Time `json:"last_round_at,omitempty"` |
| + // DeferredRounds counts the cut's final rounds that deferred an object |
| + // or moved more than the bar (§7.4 staged; three revert). LeaseUntil is |
| + // the seal's writers' bound (ruling G7), recorded beside it. |
| + DeferredRounds int `json:"deferred_rounds,omitempty"` |
| + LeaseUntil time.Time `json:"lease_until,omitempty"` |
| } |
| |
| // moveJSONPresent is the ledger's note that move.json already existed when |
| @@ -195,34 +200,58 @@ type srcRound struct { |
| replicaBytes int64 |
| } |
| |
| -// preparingRound is §7.3's source round, in order: no deployment bucket |
| -// refuses; the staging lease is taken and the claim revalidated under it; |
| -// the box-step refusals; the store binding; Touch; move.json; the replica |
| -// lane; the archive; the source report; Ready; the promotion; the |
| -// automatic cut. |
| +// preparingRound is §7.3's source round (sourceRound), then Ready, the |
| +// promotion (or the repin of a ready move whose pin a cut's admission |
| +// cleared) and the automatic cut. |
| func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error { |
| + sr, published, err := w.sourceRound(ctx, s, rec, fleet) |
| + if err != nil || published == nil { |
| + return err |
| + } |
| + // The invariant: a round promotes or requests a cut only when its own |
| + // source report passed every check and was published. Any failing |
| + // check ends the round here; a retry or an error ended it above. |
| + if !sourcePassed(published) { |
| + w.notice(rec.Host+"#ready", "moves: %s: this round's source report does not pass (S1 %v, S2 %v); no promotion and no cut", rec.Host, published.S1.OK, published.S2.OK) |
| + return nil |
| + } |
| + return w.decideReady(ctx, s, rec, provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}, sr, published) |
| +} |
| + |
| +// sourceRound is §7.3's source round, in order: no deployment bucket |
| +// refuses; the staging lease is taken and the claim revalidated under it; |
| +// the box-step refusals and the one-box rule; the store binding; Touch; |
| +// move.json; the replica lane; the archive; the source report. It returns |
| +// the round and the report it published. A round that ended before any |
| +// report (another worker holds the lease, a file changed under the |
| +// archive) returns a nil report and no error: the next pass runs it again. |
| +// The cut's `requested` runs it too (G13), and decides admission instead |
| +// of readiness. |
| +func (w *movesWorker) sourceRound(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) (*srcRound, *moves.SourceReport, error) { |
| cfg := w.cfg |
| id := rec.Move.ID |
| - g := provision.MoveGuard{ID: id, Incarnation: rec.Move.Incarnation} |
| + report := func(sr *srcRound) (*srcRound, *moves.SourceReport, error) { |
| + published, err := w.writeSourceReport(ctx, s, rec, sr) |
| + return sr, published, err |
| + } |
| // Round-5 finding 11: a lease in a per-participant bucket, or a |
| // 15-minute one, cannot fence staging writers. Without the deployment |
| // bucket there is no shared place for it, so the round refuses. |
| if cfg.StagingLease == nil { |
| - _, err := w.writeSourceReport(ctx, s, rec, &srcRound{s1: &moves.Check{Reason: s1NoDeployBucket}}) |
| - return err |
| + return report(&srcRound{s1: &moves.Check{Reason: s1NoDeployBucket}}) |
| } |
| tgt, ok := fleet.targets[rec.Move.To] |
| if !ok { |
| - return fmt.Errorf("target region %s publishes no data bucket yet. waiting", rec.Move.To) |
| + return nil, nil, fmt.Errorf("target region %s publishes no data bucket yet. waiting", rec.Move.To) |
| } |
| held, releaseLease, err := cfg.StagingLease(ctx, rec.Move.To, rec.Host) |
| if err != nil { |
| var he *lease.HeldError |
| if errors.As(err, &he) { |
| log.Printf("moves: %s: another worker holds the staging lease; this round waits", rec.Host) |
| - return nil |
| + return nil, nil, nil |
| } |
| - return fmt.Errorf("staging lease: %w", err) |
| + return nil, nil, fmt.Errorf("staging lease: %w", err) |
| } |
| defer func() { |
| rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| @@ -234,26 +263,30 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| // K5: revalidate the claim under the lease. |
| obs, err := hostclaim.Read(ctx, s, rec.Host) |
| if err != nil { |
| - return fmt.Errorf("read the host claim: %w", err) |
| + return nil, nil, fmt.Errorf("read the host claim: %w", err) |
| } |
| if obs.Absent { |
| - return fmt.Errorf("move %s has no host claim", id) |
| + return nil, nil, fmt.Errorf("move %s has no host claim", id) |
| } |
| if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil { |
| - return fmt.Errorf("move %s: %w", id, err) |
| + return nil, nil, fmt.Errorf("move %s: %w", id, err) |
| } |
| if h := obs.Claim.Holder; h == nil || h.Incarnation != rec.Move.Incarnation { |
| - return fmt.Errorf("move %s: the host claim names another incarnation", id) |
| + return nil, nil, fmt.Errorf("move %s: the host claim names another incarnation", id) |
| } |
| if cfg.Box != "" && obs.Claim.Op.Exec != cfg.Box { |
| - return fmt.Errorf("move %s: the claim's executor is %q, not this box", id, obs.Claim.Op.Exec) |
| + return nil, nil, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, obs.Claim.Op.Exec) |
| } |
| - // The box-step refusals (§7.1: every box step re-checks). |
| + // The box-step refusals (§7.1: every box step re-checks), and the |
| + // one-box rule the door applied (D15, ruling G13): a region that has |
| + // gained a second box since is refused here, every round. |
| if why, err := w.sourceRefusal(ctx, s, rec); err != nil { |
| - return err |
| + return nil, nil, err |
| } else if why != "" { |
| - _, err := w.writeSourceReport(ctx, s, rec, &srcRound{s1: &moves.Check{Reason: why}}) |
| - return err |
| + return report(&srcRound{s1: &moves.Check{Reason: why}}) |
| + } |
| + if why := oneBoxRefusal(fleet, rec); why != "" { |
| + return report(&srcRound{s1: &moves.Check{Reason: why}}) |
| } |
| // K12 and T9: the target region and box are in touched before the |
| // first staging write. |
| @@ -264,22 +297,22 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| boxes = []string{targetBox} |
| } |
| if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Touch(id, []string{rec.Move.To}, boxes)); err != nil { |
| - return fmt.Errorf("touch %s in the host claim: %w", rec.Move.To, err) |
| + return nil, nil, fmt.Errorf("touch %s in the host claim: %w", rec.Move.To, err) |
| } |
| } |
| stg, err := w.staging(ctx, tgt.DataBucketRegion) |
| if err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| own, err := w.mirror(ctx, tgt.DataBucketRegion) |
| if err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| gm := &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: cfg.Now}, dst: tgt.DataBucket, own: own} |
| ledgerKey := moves.LedgerKey(rec.Host, id) |
| ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey) |
| if err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| before := ledger |
| saveLedger := func() error { |
| @@ -298,19 +331,19 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| if ledger.MoveJSONVersion == "" { |
| body, err := json.Marshal(moves.MoveJSON{MoveID: id, Incarnation: rec.Move.Incarnation, Host: rec.Host, From: rec.Move.From, To: rec.Move.To}) |
| if err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| ver, err := gm.PutIfAbsentVersioned(ctx, tgt.DataBucket, moves.MoveJSONKey(rec.Host, id), body) |
| switch { |
| case errors.Is(err, lease.ErrExists): |
| ledger.MoveJSONVersion = moveJSONPresent |
| case err != nil: |
| - return fmt.Errorf("write move.json: %w", err) |
| + return nil, nil, fmt.Errorf("write move.json: %w", err) |
| default: |
| ledger.MoveJSONVersion = ver |
| } |
| if err := saveLedger(); err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| } |
| |
| @@ -330,13 +363,13 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| log.Printf("moves: %s: replica round failed, retrying once: %v", rec.Host, err) |
| } |
| if err != nil { |
| - return fmt.Errorf("replica lane: %w", err) |
| + return nil, nil, fmt.Errorf("replica lane: %w", err) |
| } |
| if round.BytesCopied >= crossMinBytes { |
| ledger.LastRoundBytes, ledger.LastRoundSeconds, ledger.LastRoundAt = round.BytesCopied, cfg.Now().Sub(start).Seconds(), cfg.Now().UTC() |
| } else if _, at := ledger.crossRate(); at.IsZero() || cfg.Now().Sub(at) > crossRefreshAfter { |
| if err := w.refreshCrossRate(ctx, gm, srcPrefix, keep, tgt.DataBucket, dstPrefix, &ledger); err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| } |
| |
| @@ -346,7 +379,7 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| // promotion. |
| replicaBytes, err := w.replicaLogicalSize(ctx, gm, srcPrefix, keep) |
| if err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| sr := &srcRound{deltaBytes: round.DeltaBytes(), replicaBytes: replicaBytes} |
| // The files archive, rebuilt only when its plaintext digest changes. A |
| @@ -355,30 +388,41 @@ func (w *movesWorker) preparingRound(ctx context.Context, s release.Store, rec p |
| // files, and the next round builds the one that is (review finding 2). |
| why, retry, err := w.publishArchive(ctx, gm, rec, tgt.DataBucket, &ledger) |
| if err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| if err := saveLedger(); err != nil { |
| - return err |
| + return nil, nil, err |
| } |
| if retry { |
| - return nil |
| + return nil, nil, nil |
| } |
| if why != "" { |
| sr.s1 = &moves.Check{Reason: why} |
| } |
| sr.ledger = ledger |
| - published, err := w.writeSourceReport(ctx, s, rec, sr) |
| - if err != nil { |
| - return err |
| + return report(sr) |
| +} |
| + |
| +// oneBoxRefusal is the one-box rule (D15), re-checked from the edges |
| +// snapshot every round (ruling G13): either region publishing more than one |
| +// box refuses with M-C21's sentence. It counts every box that publishes the |
| +// placement, as T8 does for the target. |
| +func oneBoxRefusal(fleet moveFleet, rec provision.Record) string { |
| + for _, region := range []string{rec.Move.To, rec.Move.From} { |
| + if len(fleet.boxes[region]) > 1 { |
| + return s1ManyBoxes(region) |
| + } |
| } |
| - // The invariant: a round promotes or requests a cut only when its own |
| - // source report passed every check and was published. Any failing |
| - // check ends the round here; a retry or an error ended it above. |
| - if !sourcePassed(published) { |
| - w.notice(rec.Host+"#ready", "moves: %s: this round's source report does not pass (S1 %v, S2 %v); no promotion and no cut", rec.Host, published.S1.OK, published.S2.OK) |
| - return nil |
| + return "" |
| +} |
| + |
| +// s1ManyBoxes is M-C21 as the source says it; the home region has no code |
| +// of its own on a box. |
| +func s1ManyBoxes(region string) string { |
| + if region == "" { |
| + region = "the home region" |
| } |
| - return w.decideReady(ctx, s, rec, g, sr, published) |
| + return region + " has more than one box. Moves between regions with several boxes are not supported yet." |
| } |
| |
| // refreshCrossRate measures the cross-region path when no recent round did: |
| @@ -510,72 +554,166 @@ func (w *movesWorker) publishArchive(ctx context.Context, gm *guardedMirror, rec |
| if plain == ledger.ArchivePlain && ledger.ArchiveVersions != nil { |
| return "", false, nil |
| } |
| - // Built and sealed to files on the data disk, and staged from the |
| - // sealed file, so a 256 MiB archive is never held in memory. |
| - tmp, err := movesScratch(cfg.DataDir, "archive-") |
| + a, why, retry, err := w.buildArchive(rec) |
| + if err != nil || why != "" || retry { |
| + return why, retry, err |
| + } |
| + ref, err := a.stage(ctx, gm, dstBucket, rec.Host, rec.Move.ID) |
| if err != nil { |
| return "", false, err |
| } |
| + ledger.ArchivePlain, ledger.ArchiveSealed, ledger.ArchiveVersions = a.Plain, a.Sealed, ref |
| + log.Printf("moves: %s: staged the files archive %s (%d files, %d bytes)", rec.Host, a.Plain, a.Entries, a.Bytes) |
| + return "", false, nil |
| +} |
| + |
| +// builtArchive is one built and sealed archive on the data disk, in the |
| +// move's archive directory (archiveDir): <Name>.sealed and <Name>.json. |
| +// The preparing round keeps the newest one there, so the cut can write the |
| +// very object the target verified into move-snapshots/ (ruling G14). |
| +type builtArchive struct { |
| + Name string `json:"name"` // files-<sealed hex> |
| + Plain string `json:"plain"` // the plaintext digest |
| + Sealed string `json:"sealed"` // the sealed object's sha256 |
| + Bytes int64 `json:"bytes"` // plaintext bytes |
| + Entries int `json:"entries"` |
| + SealedSize int64 `json:"sealed_size"` |
| + dir string |
| +} |
| + |
| +func (a builtArchive) sealedPath() string { return filepath.Join(a.dir, a.Name+".sealed") } |
| +func (a builtArchive) manifestPath() string { return filepath.Join(a.dir, a.Name+".json") } |
| + |
| +// archiveDir is where a move's built archive lives on the data disk, under |
| +// its journal directory: never a shared tmpfs. |
| +func archiveDir(dataDir, host, op string) string { |
| + return filepath.Join(moves.JournalDir(dataDir, host, op), "archive") |
| +} |
| + |
| +// archiveName is the object name for a sealed digest. |
| +func archiveName(sealed string) string { return "files-" + digestHex(sealed) } |
| + |
| +// buildArchive builds and seals the files archive into the move's archive |
| +// directory, streamed to disk so a 256 MiB archive is never held in |
| +// memory, and removes any older one there. Sealing is randomized (§8.2): |
| +// every build is a new object with a new name. It returns an S1 reason for |
| +// an archive this move can never carry, or retry for a file that changed |
| +// while it was read. |
| +func (w *movesWorker) buildArchive(rec provision.Record) (a builtArchive, why string, retry bool, err error) { |
| + cfg := w.cfg |
| + dir, envPath := cfg.FilesDir(rec.Host) |
| + adir := archiveDir(cfg.DataDir, rec.Host, rec.Move.ID) |
| + if err := os.MkdirAll(adir, 0o700); err != nil { |
| + return a, "", false, err |
| + } |
| + tmp, err := os.MkdirTemp(adir, "build-") |
| + if err != nil { |
| + return a, "", false, err |
| + } |
| defer os.RemoveAll(tmp) |
| built, err := os.Create(filepath.Join(tmp, "archive.tar.zst")) |
| if err != nil { |
| - return "", false, err |
| + return a, "", false, err |
| } |
| defer built.Close() |
| plain, entries, bytes, err := archiveBuild(dir, envPath, built) |
| if why, retry, ferr := archiveFailure(rec.Host, err); ferr != nil || why != "" || retry { |
| - return why, retry, ferr |
| + return a, why, retry, ferr |
| } |
| if _, err := built.Seek(0, io.SeekStart); err != nil { |
| - return "", false, err |
| + return a, "", false, err |
| } |
| - sealedPath := filepath.Join(tmp, "archive.sealed") |
| - sealed, err := os.Create(sealedPath) |
| + sealedTmp := filepath.Join(tmp, "archive.sealed") |
| + sealed, err := os.Create(sealedTmp) |
| if err != nil { |
| - return "", false, err |
| + return a, "", false, err |
| } |
| defer sealed.Close() |
| // Seal hashes what it writes: sealedSHA is the sealed file's sha256, |
| // the payload hash its upload is signed with. |
| dataKey, sealedSHA, err := archive.Seal(built, cfg.SecretsPub, appconfig.Binding{Account: rec.Account, App: rec.App}, sealed) |
| if err != nil { |
| - return "", false, fmt.Errorf("seal the archive: %w", err) |
| + return a, "", false, fmt.Errorf("seal the archive: %w", err) |
| } |
| if err := sealed.Sync(); err != nil { |
| - return "", false, err |
| + return a, "", false, err |
| } |
| st, err := sealed.Stat() |
| if err != nil { |
| - return "", false, err |
| + return a, "", false, err |
| } |
| m := archive.Manifest{V: archive.ManifestVersion, Op: rec.Move.ID, Incarnation: rec.Move.Incarnation, Plain: plain, |
| Sealed: sealedSHA, Bytes: bytes, Entries: entries, SealedDataKey: dataKey, Chunk: archive.ChunkSize} |
| manifest, err := json.Marshal(m) |
| if err != nil { |
| - return "", false, err |
| + return a, "", false, err |
| + } |
| + a = builtArchive{Name: archiveName(sealedSHA), Plain: plain, Sealed: sealedSHA, Bytes: bytes, Entries: len(entries), SealedSize: st.Size(), dir: adir} |
| + if err := writeFileSync(filepath.Join(tmp, "manifest.json"), manifest); err != nil { |
| + return a, "", false, err |
| + } |
| + if err := os.Rename(sealedTmp, a.sealedPath()); err != nil { |
| + return a, "", false, err |
| } |
| - name := "files-" + digestHex(sealedSHA) |
| - prefix := moves.FilesPrefix(rec.Host, rec.Move.ID) |
| - ref := &moves.ArchiveRef{ManifestKey: prefix + name + ".json", SealedObjectKey: prefix + name + ".sealed", Plain: plain} |
| - if ref.SealedVersion, err = gm.PutFileVersioned(ctx, dstBucket, ref.SealedObjectKey, sealedPath, st.Size(), digestHex(sealedSHA)); err != nil { |
| - return "", false, fmt.Errorf("stage the sealed archive: %w", err) |
| + if err := os.Rename(filepath.Join(tmp, "manifest.json"), a.manifestPath()); err != nil { |
| + return a, "", false, err |
| + } |
| + if err := syncDir(adir); err != nil { |
| + return a, "", false, err |
| + } |
| + pruneArchives(adir, a.Name) |
| + return a, "", false, nil |
| +} |
| + |
| +// stage puts the archive's two objects into the staging lane's files/ |
| +// through the guard (the sealed object streamed from disk) and returns |
| +// them by the versions the PUTs answered. |
| +func (a builtArchive) stage(ctx context.Context, gm *guardedMirror, dstBucket, host, op string) (*moves.ArchiveRef, error) { |
| + manifest, err := os.ReadFile(a.manifestPath()) |
| + if err != nil { |
| + return nil, err |
| + } |
| + prefix := moves.FilesPrefix(host, op) |
| + ref := &moves.ArchiveRef{ManifestKey: prefix + a.Name + ".json", SealedObjectKey: prefix + a.Name + ".sealed", Plain: a.Plain} |
| + if ref.SealedVersion, err = gm.PutFileVersioned(ctx, dstBucket, ref.SealedObjectKey, a.sealedPath(), a.SealedSize, digestHex(a.Sealed)); err != nil { |
| + return nil, fmt.Errorf("stage the sealed archive: %w", err) |
| } |
| if ref.ManifestVersion, err = gm.PutVersioned(ctx, dstBucket, ref.ManifestKey, manifest); err != nil { |
| - return "", false, fmt.Errorf("stage the archive manifest: %w", err) |
| + return nil, fmt.Errorf("stage the archive manifest: %w", err) |
| } |
| - ledger.ArchivePlain, ledger.ArchiveSealed, ledger.ArchiveVersions = plain, sealedSHA, ref |
| - log.Printf("moves: %s: staged the files archive %s (%d files, %d bytes)", rec.Host, plain, len(entries), bytes) |
| - return "", false, nil |
| + return ref, nil |
| } |
| |
| -// movesScratch is a scratch directory under <dataDir>/.moves: on the data |
| -// disk, never a shared tmpfs. |
| -func movesScratch(dataDir, pattern string) (string, error) { |
| - parent := filepath.Join(dataDir, ".moves") |
| - if err := os.MkdirAll(parent, 0o755); err != nil { |
| - return "", err |
| +// pruneArchives removes every archive in dir but keep. Best effort: a file |
| +// left behind costs disk until the move's directory is removed. |
| +func pruneArchives(dir, keep string) { |
| + ents, err := os.ReadDir(dir) |
| + if err != nil { |
| + return |
| + } |
| + for _, e := range ents { |
| + name := e.Name() |
| + if strings.HasPrefix(name, "files-") && strings.TrimSuffix(strings.TrimSuffix(name, ".sealed"), ".json") != keep { |
| + os.Remove(filepath.Join(dir, name)) |
| + } |
| } |
| - return os.MkdirTemp(parent, pattern) |
| +} |
| + |
| +// writeFileSync writes body to path and syncs it. |
| +func writeFileSync(path string, body []byte) error { |
| + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := f.Write(body); err != nil { |
| + f.Close() |
| + return err |
| + } |
| + if err := f.Sync(); err != nil { |
| + f.Close() |
| + return err |
| + } |
| + return f.Close() |
| } |
| |
| // archiveFailure maps an archive error to an S1 reason (a standing refusal |
| @@ -703,7 +841,10 @@ func (w *movesWorker) probeSource(ctx context.Context, s release.Store, rec prov |
| func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec provision.Record, g provision.MoveGuard, r *srcRound, source *moves.SourceReport) error { |
| cfg := w.cfg |
| state := rec.Move.State |
| - if state == provision.MovePreparing { |
| + // A ready move with no pin is one a cut's admission sent back (G1): the |
| + // same rule pins it again, from this passing round, before any cut. |
| + repin := state == provision.MoveReady && rec.Move.Inputs == nil |
| + if state == provision.MovePreparing || repin { |
| target, err := readTargetReport(ctx, s, rec) |
| if err != nil { |
| return err |
| @@ -716,16 +857,24 @@ func (w *movesWorker) decideReady(ctx context.Context, s release.Store, rec prov |
| // moves.Ready has required target.Inputs.Release to equal the |
| // source's, so what is pinned is what T7 checked (review finding 3). |
| in := target.Inputs |
| - if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedPromoteReady(g, cfg.Now(), in)); err != nil { |
| + mut := provision.MoveGuardedPromoteReady(g, cfg.Now(), in) |
| + if repin { |
| + mut = provision.MoveGuardedRepin(g, in) |
| + } |
| + if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, mut); err != nil { |
| if errors.Is(err, provision.ErrMoveConflict) { |
| - log.Printf("moves: %s: the record no longer holds move %s as preparing (%v); this round ends", rec.Host, g.ID, err) |
| + log.Printf("moves: %s: the record no longer holds move %s as %s without a pin (%v); this round ends", rec.Host, g.ID, state, err) |
| return nil |
| } |
| return err |
| } |
| - state = provision.MoveReady |
| nudgeSerial(ctx, s) |
| - log.Printf("moves: %s: move %s is ready (last round moved %d bytes; bar %d)", rec.Host, g.ID, r.deltaBytes, cfg.ReadyDelta) |
| + if repin { |
| + log.Printf("moves: %s: move %s pinned its inputs again", rec.Host, g.ID) |
| + } else { |
| + log.Printf("moves: %s: move %s is ready (last round moved %d bytes; bar %d)", rec.Host, g.ID, r.deltaBytes, cfg.ReadyDelta) |
| + } |
| + state = provision.MoveReady |
| } |
| if state == provision.MoveReady && rec.Move.Cut == "auto" { |
| if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveGuardedCut(g)); err != nil { |
| @@ -851,5 +1000,9 @@ func (w *movesWorker) deleteOwnMoveState(ctx context.Context, rec provision.Reco |
| return fmt.Errorf("delete %s: %w", k, err) |
| } |
| } |
| + // The move's local directory: its journal and the archive kept for it. |
| + if err := os.RemoveAll(moves.JournalDir(cfg.DataDir, rec.Host, rec.Move.ID)); err != nil { |
| + return fmt.Errorf("remove the move's local directory: %w", err) |
| + } |
| return nil |
| } |
amadan