| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/lease" |
| 13 | "github.com/carlosframework/platform/internal/moves" |
| 14 | "github.com/carlosframework/platform/internal/moves/archive" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | ) |
| 17 | |
| 18 | // The cut's archive (moves spec §7.4 staged, §8.3, codex 19, ruling G14): |
| 19 | // how the stopped files directory's one sealed archive is identified, |
| 20 | // published create-only into this box's move-snapshots/ and staged into the |
| 21 | // lane's files/, and how a replay finishes exactly the identity an earlier |
| 22 | // pass journalled. The cut's phase loop is in movescut.go; this file is the |
| 23 | // archive's publication and its recovery, on their own so the recovery |
| 24 | // rules can be read in one place (Task 5 fix round 1, ruling 7). |
| 25 | // |
| 26 | // Source-disk headroom. The preparing round keeps the newest built archive |
| 27 | // on the source's data disk under <JournalDir>/archive/ for the life of |
| 28 | // the move, so the cut can publish the very object the target verified |
| 29 | // (G14). That costs up to archive.Cap (256 MiB) of the source's data disk |
| 30 | // while a move is in flight. A rebuild (the stopped files differ from the |
| 31 | // verified archive) is worse for the length of one build: the previous |
| 32 | // sealed archive is still there, buildArchive writes the compressed |
| 33 | // plaintext (archive.tar.zst) to a temp directory beside it, and then |
| 34 | // seals that into the new sealed archive before pruneArchives removes the |
| 35 | // previous one. With incompressible files all three approach the payload |
| 36 | // size, so the worst case is about three times archive.Cap plus the |
| 37 | // sealing overhead. The target's T6 does not price this source-side |
| 38 | // headroom; that is a Plan 6 follow-up. |
| 39 | // cutArchiveStep is the archive's identity in the op journal, written |
| 40 | // before any PUT so a replay publishes the same object (codex 19): the |
| 41 | // archive's name and digests, the two move-snapshots/ keys it is published |
| 42 | // under, and the manifest's bytes, so a replay that finds the sealed |
| 43 | // object published and the manifest not can write the manifest without |
| 44 | // the file (ruling 4). Reuse marks the preparing archive the target |
| 45 | // verified (G14): the seal names its staged versions and nothing is staged |
| 46 | // again. |
| 47 | type cutArchiveStep struct { |
| 48 | builtArchive |
| 49 | Reuse bool `json:"reuse,omitempty"` |
| 50 | SealedKey string `json:"sealed_key"` |
| 51 | ManifestKey string `json:"manifest_key"` |
| 52 | Manifest json.RawMessage `json:"manifest"` |
| 53 | } |
| 54 | |
| 55 | // errArchiveLost: the journalled identity can be finished by nobody. The |
| 56 | // sealed file is gone from this disk and the sealed object was never |
| 57 | // published, so there is nothing to publish it from, and a new identity |
| 58 | // is not allowed once one is journalled (ruling 4). The cut reverts. |
| 59 | var errArchiveLost = errors.New("the cut's archive is lost before it was published") |
| 60 | |
| 61 | // cutArchive publishes the cut's one sealed archive (§8.3) and returns the |
| 62 | // seal's files: the sealed object and its manifest by version. |
| 63 | // |
| 64 | // Its identity is fixed before any PUT (sealing is randomized), in the op |
| 65 | // journal, and once journalled it is final: a replay finishes that |
| 66 | // identity or the cut reverts, and never reseals under a new name |
| 67 | // (ruling 4). The first run reuses the preparing archive when its |
| 68 | // plaintext digest is the stopped files' and its staged versions are pinned |
| 69 | // and its sealed file is still on this disk (G14); otherwise it builds and |
| 70 | // seals a new one. Then, from the journalled identity (publishCutArchive): |
| 71 | // - the sealed file on disk: create-only into move-snapshots/ (an existing |
| 72 | // object is this cut's earlier PUT), then staged into files/ unless it |
| 73 | // is the reused archive, whose staged versions the ledger holds; |
| 74 | // - no file: what move-snapshots/ holds decides, and only a listing that |
| 75 | // answered decides it (an error is an error, never absence). The |
| 76 | // sealed object there and the manifest not: the manifest is written |
| 77 | // from the journalled bytes. Both there: staged into files/ by a |
| 78 | // guarded server-side copy (or, reused, nothing to stage). The sealed |
| 79 | // object absent: errArchiveLost, which is a revert. |
| 80 | // |
| 81 | // why is a recorded reason the stopped files cannot be archived. |
| 82 | func (c *cutRun) cutArchive(ctx context.Context, plain string) (builtArchive, []moves.SealedObject, string, error) { |
| 83 | cfg := c.cfg() |
| 84 | host, id := c.rec.Host, c.g.ID |
| 85 | own, vs, err := c.ownStores(ctx) |
| 86 | if err != nil { |
| 87 | return builtArchive{}, nil, "", err |
| 88 | } |
| 89 | ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, moves.LedgerKey(host, id)) |
| 90 | if err != nil { |
| 91 | return builtArchive{}, nil, "", err |
| 92 | } |
| 93 | j, err := c.journal() |
| 94 | if err != nil { |
| 95 | return builtArchive{}, nil, "", err |
| 96 | } |
| 97 | adir := archiveDir(cfg.DataDir, host, id) |
| 98 | snap := moves.SnapshotsPrefix(host, id) |
| 99 | var step cutArchiveStep |
| 100 | if raw, ok := j.Steps[cutStepArchive]; ok { |
| 101 | // Journalled: final. An entry that cannot be read, or names another |
| 102 | // plaintext than the record's, is a corrupt journal, and starting |
| 103 | // over on it could publish a second identity: an error. |
| 104 | if err := json.Unmarshal([]byte(raw), &step); err != nil || step.Name != archiveName(step.Sealed) || step.Plain != plain || |
| 105 | step.SealedKey != snap+step.Name+".sealed" || step.ManifestKey != snap+step.Name+".json" || len(step.Manifest) == 0 { |
| 106 | return builtArchive{}, nil, "", fmt.Errorf("the journalled archive identity of %s is unusable (%v)", id, err) |
| 107 | } |
| 108 | step.dir = adir |
| 109 | } else { |
| 110 | reuse := ledger.ArchivePlain == plain && ledger.ArchiveVersions != nil && ledger.ArchiveVersions.Pinned() && ledger.ArchiveSealed != "" |
| 111 | if reuse { |
| 112 | kept := builtArchive{Name: archiveName(ledger.ArchiveSealed), Sealed: ledger.ArchiveSealed, Plain: plain, dir: adir} |
| 113 | keptThere, err := sealedOnDisk(kept) |
| 114 | if err != nil { |
| 115 | return builtArchive{}, nil, "", err |
| 116 | } |
| 117 | // The kept manifest is needed here, before anything is |
| 118 | // journalled: without it the verified archive is simply not |
| 119 | // reused, and a new one is built. |
| 120 | if m, ok := keptManifest(kept); ok && keptThere { |
| 121 | kept.Bytes, kept.Entries = m.Bytes, len(m.Entries) |
| 122 | if st, err := os.Stat(kept.sealedPath()); err == nil { |
| 123 | kept.SealedSize = st.Size() |
| 124 | } |
| 125 | step = cutArchiveStep{builtArchive: kept, Reuse: true} |
| 126 | } else { |
| 127 | reuse = false |
| 128 | } |
| 129 | } |
| 130 | if !reuse { |
| 131 | a, why, retry, err := c.w.buildArchive(c.rec) |
| 132 | if err != nil || why != "" { |
| 133 | return builtArchive{}, nil, why, err |
| 134 | } |
| 135 | if retry { |
| 136 | return builtArchive{}, nil, "", errors.New("a stopped file changed while it was archived") |
| 137 | } |
| 138 | if a.Plain != plain { |
| 139 | return builtArchive{}, nil, "", fmt.Errorf("the stopped files' digest is %s, recorded %s", a.Plain, plain) |
| 140 | } |
| 141 | step = cutArchiveStep{builtArchive: a} |
| 142 | } |
| 143 | manifest, err := os.ReadFile(step.manifestPath()) |
| 144 | if err != nil { |
| 145 | return builtArchive{}, nil, "", err |
| 146 | } |
| 147 | step.SealedKey, step.ManifestKey, step.Manifest = snap+step.Name+".sealed", snap+step.Name+".json", manifest |
| 148 | body, err := json.Marshal(step) |
| 149 | if err != nil { |
| 150 | return builtArchive{}, nil, "", err |
| 151 | } |
| 152 | j.Mark(cutStepArchive, string(body)) |
| 153 | if err := j.Commit(cfg.DataDir); err != nil { |
| 154 | return builtArchive{}, nil, "", err |
| 155 | } |
| 156 | if err := c.fault("archive-journaled"); err != nil { |
| 157 | return builtArchive{}, nil, "", err |
| 158 | } |
| 159 | } |
| 160 | a := step.builtArchive |
| 161 | files, err := c.publishCutArchive(ctx, own, vs, step, ledger) |
| 162 | if errors.Is(err, errArchiveLost) { |
| 163 | log.Printf("moves: %s: %v; the cut reverts", host, err) |
| 164 | return builtArchive{}, nil, provision.RevertArchiveRefused, nil |
| 165 | } |
| 166 | return a, files, "", err |
| 167 | } |
| 168 | |
| 169 | // publishCutArchive is cutArchive's PUTs from a journalled identity. The |
| 170 | // manifest it stages is the journalled one: the file on disk, when there |
| 171 | // is one, was written by the same build, and the journal is what a replay |
| 172 | // without the file has. |
| 173 | func (c *cutRun) publishCutArchive(ctx context.Context, own lease.MirrorStore, vs lease.VersionedStore, step cutArchiveStep, ledger moveLedger) ([]moves.SealedObject, error) { |
| 174 | cfg := c.cfg() |
| 175 | host, id := c.rec.Host, c.g.ID |
| 176 | a := step.builtArchive |
| 177 | snap := moves.SnapshotsPrefix(host, id) |
| 178 | sealedSnap, manifestSnap := step.SealedKey, step.ManifestKey |
| 179 | manifest := []byte(step.Manifest) |
| 180 | onDisk, err := sealedOnDisk(a) |
| 181 | if err != nil { |
| 182 | return nil, err |
| 183 | } |
| 184 | if onDisk { |
| 185 | fp, ok := vs.(lease.FilePutIfAbsent) |
| 186 | if !ok { |
| 187 | return nil, errors.New("this box cannot write the archive create-only") |
| 188 | } |
| 189 | if _, err := fp.PutFileIfAbsent(ctx, cfg.OwnBucket, sealedSnap, a.sealedPath()); err != nil && !errors.Is(err, lease.ErrExists) { |
| 190 | return nil, fmt.Errorf("snapshot the sealed archive: %w", err) |
| 191 | } |
| 192 | if err := c.fault("archive-sealed-snapshot"); err != nil { |
| 193 | return nil, err |
| 194 | } |
| 195 | if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, manifestSnap, manifest); err != nil && !errors.Is(err, lease.ErrExists) { |
| 196 | return nil, fmt.Errorf("snapshot the archive manifest: %w", err) |
| 197 | } |
| 198 | if err := c.fault("archive-own-put"); err != nil { |
| 199 | return nil, err |
| 200 | } |
| 201 | } else { |
| 202 | // No file (the read succeeded and found none, or a file that is not |
| 203 | // this archive): only what was published under the identity can |
| 204 | // finish it. A listing that failed decides nothing (ruling 4). The |
| 205 | // loss is established only here, by reads that answered. |
| 206 | sealedThere, _, err := c.snapshotState(ctx, vs, sealedSnap) |
| 207 | if err != nil { |
| 208 | return nil, err |
| 209 | } |
| 210 | if !sealedThere { |
| 211 | return nil, fmt.Errorf("%w: %s is not on this disk and not in %s", errArchiveLost, a.Name, snap) |
| 212 | } |
| 213 | manifestThere, _, err := c.snapshotState(ctx, vs, manifestSnap) |
| 214 | if err != nil { |
| 215 | return nil, err |
| 216 | } |
| 217 | if !manifestThere { |
| 218 | // A crash between the two PUTs: the manifest from the journal. |
| 219 | if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, manifestSnap, manifest); err != nil && !errors.Is(err, lease.ErrExists) { |
| 220 | return nil, fmt.Errorf("snapshot the archive manifest from the journal: %w", err) |
| 221 | } |
| 222 | log.Printf("moves: %s: the manifest of %s was written from the journalled identity", host, a.Name) |
| 223 | } |
| 224 | } |
| 225 | if step.Reuse { |
| 226 | ref := ledger.ArchiveVersions |
| 227 | if ref == nil || !ref.Pinned() || ledger.ArchiveSealed != a.Sealed { |
| 228 | return nil, fmt.Errorf("the ledger no longer names the archive %s", a.Name) |
| 229 | } |
| 230 | return []moves.SealedObject{ |
| 231 | {Key: ref.SealedObjectKey, VersionID: ref.SealedVersion, Size: a.SealedSize}, |
| 232 | {Key: ref.ManifestKey, VersionID: ref.ManifestVersion, Size: int64(len(manifest))}, |
| 233 | }, nil |
| 234 | } |
| 235 | prefix := moves.FilesPrefix(host, id) |
| 236 | sealedKey, manifestKey := prefix+a.Name+".sealed", prefix+a.Name+".json" |
| 237 | var sv, mv string |
| 238 | var msize int64 |
| 239 | if onDisk { |
| 240 | if sv, err = c.gm.PutFileVersioned(ctx, c.gm.dst, sealedKey, a.sealedPath(), a.SealedSize, digestHex(a.Sealed)); err != nil { |
| 241 | return nil, fmt.Errorf("stage the sealed archive: %w", err) |
| 242 | } |
| 243 | if err := c.fault("archive-first-copy"); err != nil { |
| 244 | return nil, err |
| 245 | } |
| 246 | if mv, err = c.gm.PutVersioned(ctx, c.gm.dst, manifestKey, manifest); err != nil { |
| 247 | return nil, fmt.Errorf("stage the archive manifest: %w", err) |
| 248 | } |
| 249 | msize = int64(len(manifest)) |
| 250 | } else { |
| 251 | // The file is gone, the snapshot is there: a server-side copy from |
| 252 | // it, guarded, as the replica lane copies. |
| 253 | objs, err := own.List(ctx, cfg.OwnBucket, snap) |
| 254 | if err != nil { |
| 255 | return nil, fmt.Errorf("list the snapshots: %w", err) |
| 256 | } |
| 257 | so, sok := objs[sealedSnap] |
| 258 | mo, mok := objs[manifestSnap] |
| 259 | if !sok || !mok { |
| 260 | return nil, fmt.Errorf("the snapshot of %s is incomplete", a.Name) |
| 261 | } |
| 262 | if _, sv, err = c.gm.CopyVersioned(ctx, cfg.OwnBucket, sealedSnap, so.ETag, c.gm.dst, sealedKey); err != nil { |
| 263 | return nil, fmt.Errorf("stage the sealed archive: %w", err) |
| 264 | } |
| 265 | if err := c.fault("archive-first-copy"); err != nil { |
| 266 | return nil, err |
| 267 | } |
| 268 | if _, mv, err = c.gm.CopyVersioned(ctx, cfg.OwnBucket, manifestSnap, mo.ETag, c.gm.dst, manifestKey); err != nil { |
| 269 | return nil, fmt.Errorf("stage the archive manifest: %w", err) |
| 270 | } |
| 271 | a.SealedSize, msize = so.Size, mo.Size |
| 272 | } |
| 273 | if err := c.fault("archive-second-copy"); err != nil { |
| 274 | return nil, err |
| 275 | } |
| 276 | if !lease.UsableVersionID(sv) || !lease.UsableVersionID(mv) { |
| 277 | return nil, fmt.Errorf("the staged archive has no usable version ids (%q, %q)", sv, mv) |
| 278 | } |
| 279 | log.Printf("moves: %s: staged the cut's archive %s", host, a.Name) |
| 280 | return []moves.SealedObject{{Key: sealedKey, VersionID: sv, Size: a.SealedSize}, {Key: manifestKey, VersionID: mv, Size: msize}}, nil |
| 281 | } |
| 282 | |
| 283 | // sealedOnDisk reports whether the archive's sealed file is on this disk |
| 284 | // and hashes to its name. The sealed file alone is what a publication |
| 285 | // needs: the manifest's bytes are journalled with the identity, so losing |
| 286 | // only the local manifest loses nothing (fix round 2). Absence, and a file |
| 287 | // whose digest is not the name, are false with no error; a read that |
| 288 | // fails is an error, never absence, since a loss is declared on this |
| 289 | // answer and a read error establishes none. |
| 290 | func sealedOnDisk(a builtArchive) (bool, error) { |
| 291 | sum, err := fileSHA256(a.sealedPath()) |
| 292 | if errors.Is(err, os.ErrNotExist) { |
| 293 | return false, nil |
| 294 | } |
| 295 | if err != nil { |
| 296 | return false, fmt.Errorf("read %s: %w", a.sealedPath(), err) |
| 297 | } |
| 298 | return sum == a.Sealed, nil |
| 299 | } |
| 300 | |
| 301 | // keptManifest reads the manifest of an archive kept on disk; it must |
| 302 | // describe that archive. |
| 303 | func keptManifest(a builtArchive) (archive.Manifest, bool) { |
| 304 | var m archive.Manifest |
| 305 | body, err := os.ReadFile(a.manifestPath()) |
| 306 | if err != nil || json.Unmarshal(body, &m) != nil || m.Sealed != a.Sealed || m.Plain != a.Plain { |
| 307 | return m, false |
| 308 | } |
| 309 | return m, true |
| 310 | } |
| 311 | |
| 312 | // snapshotState reports whether key holds a current object (not a delete |
| 313 | // marker) in this box's bucket, and its size. A listing that fails is an |
| 314 | // error, never absence (ruling 4): absence lets a replay decide the |
| 315 | // identity was never published. |
| 316 | func (c *cutRun) snapshotState(ctx context.Context, vs lease.VersionedStore, key string) (present bool, size int64, err error) { |
| 317 | vers, err := vs.ListVersions(ctx, c.cfg().OwnBucket, key) |
| 318 | if err != nil { |
| 319 | return false, 0, fmt.Errorf("list %s: %w", key, err) |
| 320 | } |
| 321 | for _, v := range vers { |
| 322 | if v.Key == key && v.IsLatest && !v.DeleteMarker { |
| 323 | return true, v.Size, nil |
| 324 | } |
| 325 | } |
| 326 | return false, 0, nil |
| 327 | } |
| 328 | |
| 329 | // sealNamesArchive reports whether the seal's files are the archive with |
| 330 | // the sealed digest given. |
| 331 | func sealNamesArchive(seal moves.Seal, sealed string) bool { |
| 332 | if sealed == "" { |
| 333 | return false |
| 334 | } |
| 335 | name := archiveName(sealed) |
| 336 | n := 0 |
| 337 | for _, o := range seal.Files { |
| 338 | base := filepath.Base(o.Key) |
| 339 | if base == name+".sealed" || base == name+".json" { |
| 340 | n++ |
| 341 | } |
| 342 | } |
| 343 | return n == 2 && len(seal.Files) == 2 |
| 344 | } |
| 345 | |
amadan