| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "net/url" |
| 11 | "sort" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | |
| 15 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 16 | s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" |
| 17 | "github.com/aws/smithy-go" |
| 18 | ) |
| 19 | |
| 20 | // The staging mirror (issue #331, instance-move spec): a residence move |
| 21 | // copies the instance's REPLICA prefix — the litestream generations the |
| 22 | // activator already maintains — never the database itself. MirrorPrefix is |
| 23 | // the one delta round; the move worker calls it repeatedly until a round |
| 24 | // copies little enough that the cut's own round is cheap, and once more |
| 25 | // after the child is down. |
| 26 | // |
| 27 | // The interface exists for the tests (an in-memory fake) and because the |
| 28 | // destination bucket lives in ANOTHER region: the implementation holds one |
| 29 | // client per side. |
| 30 | |
| 31 | // MirrorStore is the S3 actions a delta round needs. List answers |
| 32 | // key→object under a prefix (keys RELATIVE to the bucket root, not the |
| 33 | // prefix). Copy is a server-side CopyObject, conditional on srcETag when it |
| 34 | // is set: a source that no longer carries that ETag answers |
| 35 | // ErrCopySourceChanged and copies nothing. It returns the ETag the |
| 36 | // destination object landed with. Delete removes one key. Get and Put read |
| 37 | // and write one small whole object (the copy receipts); Get answers |
| 38 | // ErrMirrorNotFound for a missing key. |
| 39 | type MirrorStore interface { |
| 40 | List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) |
| 41 | Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (dstETag string, err error) |
| 42 | Delete(ctx context.Context, bucket, key string) error |
| 43 | Get(ctx context.Context, bucket, key string) ([]byte, error) |
| 44 | Put(ctx context.Context, bucket, key string, body []byte) error |
| 45 | } |
| 46 | |
| 47 | // VersionedCopier is the optional half of MirrorStore a versioned bucket |
| 48 | // supports: the copy's response carries the destination's version id, |
| 49 | // which a move's seal records (moves spec §7.3, "seal by version"). The |
| 50 | // source never lists the target's staging, so the response is the only |
| 51 | // place the id can come from. Only the staging mirror (Task 6, |
| 52 | // lease.NewStagingMirror, through Task 14's guardedMirror) implements it; |
| 53 | // the ordinary s3Mirror and every plain MirrorStore record no versions and |
| 54 | // MirrorRound.Versions stays empty, so production mirroring (NewS3Mirror, |
| 55 | // the legacy worker) is unchanged (round-4 finding 7). |
| 56 | type VersionedCopier interface { |
| 57 | CopyVersioned(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (etag, versionID string, err error) |
| 58 | } |
| 59 | |
| 60 | // VersionedPutter is the same for Put: the archive, the timers snapshot |
| 61 | // and move.json are sealed by the version their PUT returned. |
| 62 | type VersionedPutter interface { |
| 63 | PutVersioned(ctx context.Context, bucket, key string, body []byte) (versionID string, err error) |
| 64 | } |
| 65 | |
| 66 | // ErrCopySourceChanged is a conditional Copy whose source was overwritten |
| 67 | // after the round listed it. The round defers that object to the next |
| 68 | // round rather than failing: a source that keeps changing is a busy |
| 69 | // source, not a broken one. |
| 70 | var ErrCopySourceChanged = errors.New("mirror: copy source changed since it was listed") |
| 71 | |
| 72 | // ErrMirrorNotFound is Get's answer for a key that does not exist. |
| 73 | var ErrMirrorNotFound = errors.New("mirror: no such object") |
| 74 | |
| 75 | // MirrorObject is one listed object: the ETag the sameness test compares, |
| 76 | // and the size a round's byte accounting adds up. |
| 77 | type MirrorObject struct { |
| 78 | ETag string |
| 79 | Size int64 |
| 80 | } |
| 81 | |
| 82 | // MirrorRound is what one delta round did. |
| 83 | type MirrorRound struct { |
| 84 | // Changed counts objects copied, deferred or deleted; 0 means the |
| 85 | // sides were identical when listed. |
| 86 | Changed int |
| 87 | // BytesCopied sums the listed SOURCE size of every object copied. |
| 88 | BytesCopied int64 |
| 89 | // Deferred counts objects whose conditional copy found the source |
| 90 | // overwritten since the listing (ErrCopySourceChanged); BytesDeferred |
| 91 | // sums their listed sizes. They are still to copy, so a round that |
| 92 | // has to be byte-final (the cut's) must not count as done while |
| 93 | // Deferred > 0. |
| 94 | Deferred int |
| 95 | BytesDeferred int64 |
| 96 | // Versions maps a relative key to the destination version id of the |
| 97 | // copy that produced its current destination object, for every current |
| 98 | // source key whose version is known (a copy this round, or a receipt |
| 99 | // that still vouches for the destination). Non-nil only when the store |
| 100 | // is a VersionedCopier; a plain MirrorStore records none. |
| 101 | Versions map[string]string |
| 102 | } |
| 103 | |
| 104 | // DeltaBytes is how much the round found to move: copied plus deferred. |
| 105 | // The move worker's readiness bar measures this. An awake instance never |
| 106 | // reaches Changed == 0 (litestream writes new files and compaction |
| 107 | // rewrites old ones between any two listings), but the bytes a round had |
| 108 | // to move bound the work left for the cut's round. |
| 109 | func (r MirrorRound) DeltaBytes() int64 { return r.BytesCopied + r.BytesDeferred } |
| 110 | |
| 111 | // Receipts names the object where a mirror keeps its copy receipts. The |
| 112 | // zero value keeps none, and every object whose ETags differ is copied. |
| 113 | // |
| 114 | // Why receipts exist (#175): S3's CopyObject always writes a single-part |
| 115 | // object, so copying a MULTIPART source (litestream uploads anything over |
| 116 | // one 5 MiB part that way) lands a destination whose ETag differs from the |
| 117 | // source's. ETag equality then reports every multipart object as changed |
| 118 | // on every round, forever: production recopied six 138 MB level files a |
| 119 | // round. A receipt records what one copy did: the source ETag and size it |
| 120 | // copied, and the ETag the destination landed with. An object is skipped |
| 121 | // when the source still lists that ETag and size AND the destination still |
| 122 | // lists that destination ETag, and copied otherwise. |
| 123 | // |
| 124 | // The receipts object must live where the mirroring box can READ it. A |
| 125 | // move's destination is push-only to the source box (it may list, put and |
| 126 | // delete under moves/*, never get), so the move worker keeps them in the |
| 127 | // SOURCE's own bucket, outside every prefix a restore or a promotion |
| 128 | // reads. They are never data. |
| 129 | type Receipts struct { |
| 130 | Bucket, Key string |
| 131 | } |
| 132 | |
| 133 | // receiptBook is the receipts object's shape. It names the pair it |
| 134 | // describes, and a book written for another pair is not evidence about |
| 135 | // this one. |
| 136 | type receiptBook struct { |
| 137 | V int `json:"v"` |
| 138 | SrcBucket string `json:"src_bucket"` |
| 139 | SrcPrefix string `json:"src_prefix"` |
| 140 | DstBucket string `json:"dst_bucket"` |
| 141 | DstPrefix string `json:"dst_prefix"` |
| 142 | Objects map[string]copyReceipt `json:"objects"` // relative key → receipt |
| 143 | } |
| 144 | |
| 145 | // copyReceipt is one completed copy whose destination ETag differs from |
| 146 | // its source's, or (on a versioned store) one whose version id is known. |
| 147 | // Otherwise, where the ETags already match, the plain equality test needs |
| 148 | // no help. |
| 149 | type copyReceipt struct { |
| 150 | SrcETag string `json:"src_etag"` |
| 151 | SrcSize int64 `json:"src_size"` |
| 152 | DstETag string `json:"dst_etag"` |
| 153 | // DstVersionID is the destination version the copy returned. Written |
| 154 | // only by a VersionedCopier; a versioned round skips an object only |
| 155 | // when a receipt carrying one still vouches for it. |
| 156 | DstVersionID string `json:"dst_version_id,omitempty"` |
| 157 | } |
| 158 | |
| 159 | // MirrorPrefix makes dstBucket/dstPrefix an exact mirror of |
| 160 | // srcBucket/srcPrefix: copy keys that are new or changed, delete keys the |
| 161 | // source no longer has (litestream retention must hold in staging too: a |
| 162 | // restore that replays pruned WAL segments is a restore nobody tested). |
| 163 | // Returns what the round did; on error, what it did before the failure. |
| 164 | // |
| 165 | // "Unchanged" is exact, never a guess from size: either the two listed |
| 166 | // ETags are equal (a single-part source, whose copy keeps its ETag), or a |
| 167 | // receipt (see Receipts) proves the destination is this mirror's copy of |
| 168 | // exactly the source version now listed. |
| 169 | // |
| 170 | // Crash safety: a receipt is written only AFTER the copy it describes |
| 171 | // returned, carrying the ETag that copy returned, once per round at the |
| 172 | // end. A crash before that write loses receipts, and the next round copies |
| 173 | // those objects once more. It can never produce a wrong skip, because a |
| 174 | // skip needs the destination to still list the ETag the copy produced. |
| 175 | func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string, rc Receipts) (MirrorRound, error) { |
| 176 | return MirrorPrefixKeep(ctx, m, srcBucket, srcPrefix, dstBucket, dstPrefix, rc, nil) |
| 177 | } |
| 178 | |
| 179 | // MirrorPrefixKeep is MirrorPrefix over the source keys keep admits: keep |
| 180 | // is asked each source key relative to srcPrefix, and a key it refuses is |
| 181 | // treated as absent, so it is never copied and a destination copy of it is |
| 182 | // deleted. A move's lane (moves.ReplicaSource) mirrors one generation's |
| 183 | // LTX files and nothing else this way. A nil keep admits every key, which |
| 184 | // is MirrorPrefix. |
| 185 | func MirrorPrefixKeep(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string, rc Receipts, keep func(rel string) bool) (MirrorRound, error) { |
| 186 | // Both prefixes are DIRECTORY boundaries, normalized here to end in |
| 187 | // "/": activator.HibernatePrefix carries none (path.Join), MovesPrefix |
| 188 | // carries one, and relative-key math over a slashless prefix would both |
| 189 | // mint fused keys (instances/hgenerations/…) and let a listing bleed |
| 190 | // into a sibling host whose name merely extends this one's. |
| 191 | if !strings.HasSuffix(srcPrefix, "/") { |
| 192 | srcPrefix += "/" |
| 193 | } |
| 194 | if !strings.HasSuffix(dstPrefix, "/") { |
| 195 | dstPrefix += "/" |
| 196 | } |
| 197 | src, err := m.List(ctx, srcBucket, srcPrefix) |
| 198 | if err != nil { |
| 199 | return MirrorRound{}, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err) |
| 200 | } |
| 201 | if keep != nil { |
| 202 | for k := range src { |
| 203 | if !keep(k[len(srcPrefix):]) { |
| 204 | delete(src, k) |
| 205 | } |
| 206 | } |
| 207 | } |
| 208 | dst, err := m.List(ctx, dstBucket, dstPrefix) |
| 209 | if err != nil { |
| 210 | return MirrorRound{}, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| 211 | } |
| 212 | book := receiptLedger{m: m, rc: rc, want: receiptBook{V: 1, SrcBucket: srcBucket, SrcPrefix: srcPrefix, DstBucket: dstBucket, DstPrefix: dstPrefix}} |
| 213 | // Deterministic order: a failed round resumes the same way everywhere. |
| 214 | rel := make([]string, 0, len(src)) |
| 215 | for k := range src { |
| 216 | rel = append(rel, k[len(srcPrefix):]) |
| 217 | } |
| 218 | sort.Strings(rel) |
| 219 | var round MirrorRound |
| 220 | // Version collection is decided once, by the store: a VersionedCopier |
| 221 | // is in receipt-required mode, any other store keeps the plain rules. |
| 222 | vc, versioned := m.(VersionedCopier) |
| 223 | if versioned { |
| 224 | round.Versions = map[string]string{} |
| 225 | } |
| 226 | // The plan: which objects this round copies. Deciding a skip reads only |
| 227 | // the two listings and the receipts, so it is done here, in key order, |
| 228 | // before any copy starts; the copies then run side by side. |
| 229 | var jobs []string |
| 230 | for _, r := range rel { |
| 231 | // The lease's lock.json is COORDINATION state, not data (live |
| 232 | // rehearsal catch, 2026-08-29): the renewer rewrites it on every |
| 233 | // TTL renewal, so mirroring it would put a change in every round |
| 234 | // for any awake instance — and promoting it would hand the target a |
| 235 | // stale foreign lease. Skipped on both sides: never copied, and |
| 236 | // the destination's own copy (if any) is never deleted. |
| 237 | if r == leaseLockName { |
| 238 | continue |
| 239 | } |
| 240 | s := src[srcPrefix+r] |
| 241 | d, staged := dst[dstPrefix+r] |
| 242 | if staged && d.ETag == s.ETag && !versioned { |
| 243 | continue |
| 244 | } |
| 245 | if staged { |
| 246 | rcpt, ok, err := book.get(ctx, r) |
| 247 | if err != nil { |
| 248 | return round, err |
| 249 | } |
| 250 | // A versioned round must be able to name the version it would |
| 251 | // seal, so a matching receipt has to carry one; without it the |
| 252 | // object is recopied (a fresh version id; the stale version |
| 253 | // stays unreferenced). |
| 254 | // A receipt naming "null" (UsableVersionID) is no better than |
| 255 | // none: a later write overwrote that version. |
| 256 | if ok && rcpt.SrcETag == s.ETag && rcpt.SrcSize == s.Size && rcpt.DstETag == d.ETag && d.Size == s.Size && (!versioned || UsableVersionID(rcpt.DstVersionID)) { |
| 257 | if versioned { |
| 258 | round.Versions[r] = rcpt.DstVersionID |
| 259 | } |
| 260 | continue |
| 261 | } |
| 262 | } |
| 263 | jobs = append(jobs, r) |
| 264 | } |
| 265 | // The copies, up to mirrorWorkers(m) at once (#175 offline-move |
| 266 | // speedup: one server-side copy at a time took six minutes for a 1.6 GB |
| 267 | // replica while the instance was offline). Every update to round and |
| 268 | // to the book is made under mu, in whatever order the copies finish; |
| 269 | // neither depends on that order. The first failure stops the round: no |
| 270 | // further copy starts, the ones in flight are cancelled, and every copy |
| 271 | // that did land keeps its receipt (the best-effort save below). |
| 272 | // |
| 273 | // No network request is ever made under mu (Astra round 1, finding 3): |
| 274 | // a receipts read that stalled there would hold every other worker, |
| 275 | // the first failure's included, so the round could not stop. The book |
| 276 | // is read here, before any copy, whenever this round copies anything |
| 277 | // and keeps receipts; set and drop then touch memory only. (A round |
| 278 | // that copies nothing still reads no receipts.) |
| 279 | if len(jobs) > 0 { |
| 280 | if err := book.load(ctx); err != nil { |
| 281 | return round, err |
| 282 | } |
| 283 | } |
| 284 | var mu sync.Mutex |
| 285 | err = forEachParallel(ctx, mirrorWorkers(m), len(jobs), func(cctx context.Context, i int) error { |
| 286 | r := jobs[i] |
| 287 | sk, dk := srcPrefix+r, dstPrefix+r |
| 288 | s := src[sk] |
| 289 | var etag, ver string |
| 290 | var err error |
| 291 | if versioned { |
| 292 | etag, ver, err = vc.CopyVersioned(cctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| 293 | } else { |
| 294 | etag, err = m.Copy(cctx, srcBucket, sk, s.ETag, dstBucket, dk) |
| 295 | } |
| 296 | mu.Lock() |
| 297 | defer mu.Unlock() |
| 298 | if errors.Is(err, ErrCopySourceChanged) { |
| 299 | book.drop(r) |
| 300 | round.Changed++ |
| 301 | round.Deferred++ |
| 302 | round.BytesDeferred += s.Size |
| 303 | return nil |
| 304 | } |
| 305 | if err != nil { |
| 306 | return fmt.Errorf("mirror: copy %s: %w", sk, err) |
| 307 | } |
| 308 | if !UsableVersionID(ver) { |
| 309 | ver = "" // never recorded or sealed by |
| 310 | } |
| 311 | round.Changed++ |
| 312 | round.BytesCopied += s.Size |
| 313 | if ver != "" { |
| 314 | round.Versions[r] = ver |
| 315 | } |
| 316 | if ver != "" || (etag != "" && etag != s.ETag) { |
| 317 | dstETag := etag |
| 318 | if dstETag == "" { |
| 319 | dstETag = s.ETag |
| 320 | } |
| 321 | // The book was read before the copies began, so this is a |
| 322 | // write to memory, never a request under mu. |
| 323 | return book.set(ctx, r, copyReceipt{SrcETag: s.ETag, SrcSize: s.Size, DstETag: dstETag, DstVersionID: ver}) |
| 324 | } |
| 325 | book.drop(r) |
| 326 | return nil |
| 327 | }) |
| 328 | if err != nil { |
| 329 | book.save(ctx) // best effort: what already landed keeps its receipts |
| 330 | return round, err |
| 331 | } |
| 332 | // The deletes, only once every copy has landed (as before), and side |
| 333 | // by side among themselves: their order never mattered. |
| 334 | var gone []string |
| 335 | for dk := range dst { |
| 336 | r := dk[len(dstPrefix):] |
| 337 | if r == leaseLockName { |
| 338 | continue |
| 339 | } |
| 340 | if _, ok := src[srcPrefix+r]; !ok { |
| 341 | gone = append(gone, r) |
| 342 | } |
| 343 | } |
| 344 | sort.Strings(gone) |
| 345 | err = forEachParallel(ctx, mirrorWorkers(m), len(gone), func(cctx context.Context, i int) error { |
| 346 | r := gone[i] |
| 347 | if err := m.Delete(cctx, dstBucket, dstPrefix+r); err != nil { |
| 348 | return fmt.Errorf("mirror: delete %s: %w", dstPrefix+r, err) |
| 349 | } |
| 350 | mu.Lock() |
| 351 | defer mu.Unlock() |
| 352 | book.drop(r) |
| 353 | round.Changed++ |
| 354 | return nil |
| 355 | }) |
| 356 | if err != nil { |
| 357 | book.save(ctx) |
| 358 | return round, err |
| 359 | } |
| 360 | book.prune(func(r string) bool { _, ok := src[srcPrefix+r]; return ok }) |
| 361 | if err := book.save(ctx); err != nil { |
| 362 | return round, err |
| 363 | } |
| 364 | return round, nil |
| 365 | } |
| 366 | |
| 367 | // DefaultMirrorWorkers is how many copies (and, after them, deletes) one |
| 368 | // delta round runs at once. Each is one server-side request, so the round's |
| 369 | // wall time is the slowest request of each batch rather than their sum. |
| 370 | const DefaultMirrorWorkers = 8 |
| 371 | |
| 372 | // MirrorConcurrency is the optional half of a MirrorStore that sets how many |
| 373 | // requests a round runs at once. A store that does not implement it gets |
| 374 | // DefaultMirrorWorkers; an answer below one is one (a strictly serial |
| 375 | // round, as every round was before). Tests use it; production stores do not. |
| 376 | type MirrorConcurrency interface { |
| 377 | MirrorWorkers() int |
| 378 | } |
| 379 | |
| 380 | func mirrorWorkers(m MirrorStore) int { |
| 381 | if c, ok := m.(MirrorConcurrency); ok { |
| 382 | if n := c.MirrorWorkers(); n >= 1 { |
| 383 | return n |
| 384 | } |
| 385 | return 1 |
| 386 | } |
| 387 | return DefaultMirrorWorkers |
| 388 | } |
| 389 | |
| 390 | // forEachParallel calls do for every index below n, on up to workers |
| 391 | // goroutines, starting them in index order. The first error do returns |
| 392 | // stops it: no further index starts, the context handed to the calls still |
| 393 | // in flight is cancelled, and once they have returned that first error is |
| 394 | // the answer (any error a cancelled call returns after it is dropped). A ctx |
| 395 | // that ends before every index has started is its own error. Calls that |
| 396 | // were in flight when it stopped have returned by the time it does, so the |
| 397 | // caller's state is final. |
| 398 | func forEachParallel(ctx context.Context, workers, n int, do func(ctx context.Context, i int) error) error { |
| 399 | if n == 0 { |
| 400 | return nil |
| 401 | } |
| 402 | if workers < 1 { |
| 403 | workers = 1 |
| 404 | } |
| 405 | if workers > n { |
| 406 | workers = n |
| 407 | } |
| 408 | cctx, cancel := context.WithCancel(ctx) |
| 409 | defer cancel() |
| 410 | var ( |
| 411 | mu sync.Mutex |
| 412 | first error |
| 413 | wg sync.WaitGroup |
| 414 | ) |
| 415 | next := make(chan int) |
| 416 | for w := 0; w < workers; w++ { |
| 417 | wg.Add(1) |
| 418 | go func() { |
| 419 | defer wg.Done() |
| 420 | for i := range next { |
| 421 | if err := do(cctx, i); err != nil { |
| 422 | mu.Lock() |
| 423 | if first == nil { |
| 424 | first = err |
| 425 | cancel() |
| 426 | } |
| 427 | mu.Unlock() |
| 428 | } |
| 429 | } |
| 430 | }() |
| 431 | } |
| 432 | started := 0 |
| 433 | feed: |
| 434 | for ; started < n; started++ { |
| 435 | // Checked first: a select with both cases ready picks at random, |
| 436 | // and a stopped round must not start another copy. |
| 437 | if cctx.Err() != nil { |
| 438 | break |
| 439 | } |
| 440 | select { |
| 441 | case next <- started: |
| 442 | case <-cctx.Done(): |
| 443 | break feed |
| 444 | } |
| 445 | } |
| 446 | close(next) |
| 447 | wg.Wait() |
| 448 | if first != nil { |
| 449 | return first |
| 450 | } |
| 451 | if started < n { |
| 452 | if err := ctx.Err(); err != nil { |
| 453 | return fmt.Errorf("mirror: %w", err) |
| 454 | } |
| 455 | } |
| 456 | return nil |
| 457 | } |
| 458 | |
| 459 | // receiptLedger is one round's view of the receipts object. It reads the |
| 460 | // object lazily, on the first object whose ETags differ, so a round over |
| 461 | // single-part objects only costs no request it did not cost before. |
| 462 | type receiptLedger struct { |
| 463 | m MirrorStore |
| 464 | rc Receipts |
| 465 | want receiptBook // the pair this round mirrors; Objects unused |
| 466 | loaded bool |
| 467 | book map[string]copyReceipt |
| 468 | dirty bool |
| 469 | } |
| 470 | |
| 471 | func (l *receiptLedger) load(ctx context.Context) error { |
| 472 | if l.loaded || l.rc.Key == "" { |
| 473 | return nil |
| 474 | } |
| 475 | body, err := l.m.Get(ctx, l.rc.Bucket, l.rc.Key) |
| 476 | if err != nil && !errors.Is(err, ErrMirrorNotFound) { |
| 477 | // Not loaded: a failed read must never be saved over as if the |
| 478 | // book were empty. |
| 479 | return fmt.Errorf("mirror: read receipts %s/%s: %w", l.rc.Bucket, l.rc.Key, err) |
| 480 | } |
| 481 | l.loaded = true |
| 482 | l.book = map[string]copyReceipt{} |
| 483 | if err != nil { |
| 484 | return nil // no receipts yet |
| 485 | } |
| 486 | var b receiptBook |
| 487 | if json.Unmarshal(body, &b) != nil || b.V != 1 || b.SrcBucket != l.want.SrcBucket || b.SrcPrefix != l.want.SrcPrefix || |
| 488 | b.DstBucket != l.want.DstBucket || b.DstPrefix != l.want.DstPrefix { |
| 489 | // Unreadable, or written for another pair: no evidence at all. |
| 490 | // Rewritten at the end of the round. |
| 491 | l.dirty = true |
| 492 | return nil |
| 493 | } |
| 494 | for k, v := range b.Objects { |
| 495 | l.book[k] = v |
| 496 | } |
| 497 | return nil |
| 498 | } |
| 499 | |
| 500 | func (l *receiptLedger) get(ctx context.Context, rel string) (copyReceipt, bool, error) { |
| 501 | if err := l.load(ctx); err != nil { |
| 502 | return copyReceipt{}, false, err |
| 503 | } |
| 504 | r, ok := l.book[rel] |
| 505 | return r, ok, nil |
| 506 | } |
| 507 | |
| 508 | // set records a completed copy. It reads the book first if this round |
| 509 | // has not (a copy to a key the destination lacked needs no lookup), so the |
| 510 | // write at the end of the round keeps every other receipt. |
| 511 | func (l *receiptLedger) set(ctx context.Context, rel string, r copyReceipt) error { |
| 512 | if l.rc.Key == "" { |
| 513 | return nil |
| 514 | } |
| 515 | if err := l.load(ctx); err != nil { |
| 516 | return err |
| 517 | } |
| 518 | if old, ok := l.book[rel]; !ok || old != r { |
| 519 | l.book[rel] = r |
| 520 | l.dirty = true |
| 521 | } |
| 522 | return nil |
| 523 | } |
| 524 | |
| 525 | // drop forgets rel's receipt. Only a book already read can hold one. |
| 526 | func (l *receiptLedger) drop(rel string) { |
| 527 | if !l.loaded { |
| 528 | return |
| 529 | } |
| 530 | if _, ok := l.book[rel]; ok { |
| 531 | delete(l.book, rel) |
| 532 | l.dirty = true |
| 533 | } |
| 534 | } |
| 535 | |
| 536 | func (l *receiptLedger) prune(keep func(rel string) bool) { |
| 537 | if !l.loaded { |
| 538 | return |
| 539 | } |
| 540 | for rel := range l.book { |
| 541 | if !keep(rel) { |
| 542 | delete(l.book, rel) |
| 543 | l.dirty = true |
| 544 | } |
| 545 | } |
| 546 | } |
| 547 | |
| 548 | func (l *receiptLedger) save(ctx context.Context) error { |
| 549 | if !l.dirty || l.rc.Key == "" { |
| 550 | return nil |
| 551 | } |
| 552 | b := l.want |
| 553 | b.Objects = l.book |
| 554 | body, err := json.Marshal(b) |
| 555 | if err != nil { |
| 556 | return err |
| 557 | } |
| 558 | if err := l.m.Put(ctx, l.rc.Bucket, l.rc.Key, body); err != nil { |
| 559 | return fmt.Errorf("mirror: write receipts %s/%s: %w", l.rc.Bucket, l.rc.Key, err) |
| 560 | } |
| 561 | l.dirty = false |
| 562 | return nil |
| 563 | } |
| 564 | |
| 565 | // leaseLockName is the coordination file this package's own leaser keeps |
| 566 | // beside every replica. MirrorPrefix's exclusion of it is lease POLICY, not |
| 567 | // generic mirroring — which is why the mirror lives in this package. |
| 568 | const leaseLockName = "lock.json" |
| 569 | |
| 570 | // DeletePrefix removes every object under prefix (the lease lock included: |
| 571 | // a deleted staging area holds nothing back). The two callers are abort's |
| 572 | // box half (errors surface — the abort retries) and the activator's |
| 573 | // post-restore cleanup (best-effort at the call site). |
| 574 | func DeletePrefix(ctx context.Context, m MirrorStore, bucket, prefix string) error { |
| 575 | objs, err := m.List(ctx, bucket, prefix) |
| 576 | if err != nil { |
| 577 | return err |
| 578 | } |
| 579 | for k := range objs { |
| 580 | if err := m.Delete(ctx, bucket, k); err != nil { |
| 581 | return err |
| 582 | } |
| 583 | } |
| 584 | return nil |
| 585 | } |
| 586 | |
| 587 | // s3Mirror implements MirrorStore over two regional clients: reads and the |
| 588 | // copy call go to the DESTINATION region's client (CopyObject executes in |
| 589 | // the destination bucket's region and pulls from the source), source |
| 590 | // listings to the source's. |
| 591 | type s3Mirror struct { |
| 592 | src, dst *awss3.Client |
| 593 | // byBucket memoizes which client answered for a bucket, so only the |
| 594 | // FIRST List of each bucket pays the wrong-region probe — and a later |
| 595 | // transient error on the right client is not retried against a region |
| 596 | // that can only answer with a redirect. |
| 597 | mu sync.Mutex |
| 598 | byBucket map[string]*awss3.Client |
| 599 | } |
| 600 | |
| 601 | // NewS3Mirror builds the two-region pair with the same credential chain and |
| 602 | // endpoint compatibility every other S3 consumer here uses (endpoint.go's |
| 603 | // newLeaserClient). |
| 604 | func NewS3Mirror(ctx context.Context, srcRegion, dstRegion string) (MirrorStore, error) { |
| 605 | src, _, err := newLeaserClient(ctx, srcRegion) |
| 606 | if err != nil { |
| 607 | return nil, err |
| 608 | } |
| 609 | dst, _, err := newLeaserClient(ctx, dstRegion) |
| 610 | if err != nil { |
| 611 | return nil, err |
| 612 | } |
| 613 | return &s3Mirror{src: src, dst: dst, byBucket: map[string]*awss3.Client{}}, nil |
| 614 | } |
| 615 | |
| 616 | func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]MirrorObject, error) { |
| 617 | out := map[string]MirrorObject{} |
| 618 | pager := awss3.NewListObjectsV2Paginator(c, &awss3.ListObjectsV2Input{Bucket: &bucket, Prefix: &prefix}) |
| 619 | for pager.HasMorePages() { |
| 620 | resp, err := pager.NextPage(ctx) |
| 621 | if err != nil { |
| 622 | return nil, err |
| 623 | } |
| 624 | for _, o := range resp.Contents { |
| 625 | if o.Key == nil { |
| 626 | continue |
| 627 | } |
| 628 | var obj MirrorObject |
| 629 | if o.ETag != nil { |
| 630 | obj.ETag = normalizeETag(*o.ETag) |
| 631 | } |
| 632 | if o.Size != nil { |
| 633 | obj.Size = *o.Size |
| 634 | } |
| 635 | out[*o.Key] = obj |
| 636 | } |
| 637 | } |
| 638 | return out, nil |
| 639 | } |
| 640 | |
| 641 | // List routes by memoized bucket→client binding: the interface's caller |
| 642 | // never says which region a bucket lives in, so the FIRST touch of a bucket |
| 643 | // probes source-then-destination (a wrong-region List is an error, so the |
| 644 | // probe is unambiguous) and the winner is remembered. |
| 645 | func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]MirrorObject, error) { |
| 646 | m.mu.Lock() |
| 647 | c := m.byBucket[bucket] |
| 648 | m.mu.Unlock() |
| 649 | if c != nil { |
| 650 | return m.list(ctx, c, bucket, prefix) |
| 651 | } |
| 652 | out, err := m.list(ctx, m.src, bucket, prefix) |
| 653 | c = m.src |
| 654 | if err != nil { |
| 655 | if out, err = m.list(ctx, m.dst, bucket, prefix); err != nil { |
| 656 | return nil, err |
| 657 | } |
| 658 | c = m.dst |
| 659 | } |
| 660 | m.mu.Lock() |
| 661 | m.byBucket[bucket] = c |
| 662 | m.mu.Unlock() |
| 663 | return out, nil |
| 664 | } |
| 665 | |
| 666 | // normalizeETag strips the quotes S3 wraps an ETag in, so a listing and a |
| 667 | // CopyObject result compare equal when they name the same object: a |
| 668 | // receipt whose quoting differed from the next listing's would never match, |
| 669 | // and the recopy-forever bug would come back silently. |
| 670 | func normalizeETag(et string) string { return strings.Trim(et, `"`) } |
| 671 | |
| 672 | // clientFor is the client List bound bucket to, or fallback if this pair |
| 673 | // has never listed it. |
| 674 | func (m *s3Mirror) clientFor(bucket string, fallback *awss3.Client) *awss3.Client { |
| 675 | m.mu.Lock() |
| 676 | defer m.mu.Unlock() |
| 677 | if c := m.byBucket[bucket]; c != nil { |
| 678 | return c |
| 679 | } |
| 680 | return fallback |
| 681 | } |
| 682 | |
| 683 | func (m *s3Mirror) Copy(ctx context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) { |
| 684 | source := url.PathEscape(srcBucket + "/" + srcKey) |
| 685 | in := &awss3.CopyObjectInput{Bucket: &dstBucket, Key: &dstKey, CopySource: &source} |
| 686 | if srcETag != "" { |
| 687 | quoted := `"` + normalizeETag(srcETag) + `"` |
| 688 | in.CopySourceIfMatch = "ed |
| 689 | } |
| 690 | out, err := m.dst.CopyObject(ctx, in) |
| 691 | if err != nil { |
| 692 | var ae smithy.APIError |
| 693 | if srcETag != "" && errors.As(err, &ae) && ae.ErrorCode() == "PreconditionFailed" { |
| 694 | return "", ErrCopySourceChanged |
| 695 | } |
| 696 | return "", err |
| 697 | } |
| 698 | if out.CopyObjectResult == nil || out.CopyObjectResult.ETag == nil { |
| 699 | return "", nil // no receipt possible; the next round recopies |
| 700 | } |
| 701 | return normalizeETag(*out.CopyObjectResult.ETag), nil |
| 702 | } |
| 703 | |
| 704 | // Delete goes to the client List bound the bucket to: the receipts object |
| 705 | // lives in the SOURCE's bucket, which the destination's client cannot |
| 706 | // reach. A bucket this pair never listed (an abort after an agent restart |
| 707 | // lists only the target) tries the destination client, as before, then |
| 708 | // the source's; if both fail, the destination's error is the one reported. |
| 709 | func (m *s3Mirror) Delete(ctx context.Context, bucket, key string) error { |
| 710 | in := &awss3.DeleteObjectInput{Bucket: &bucket, Key: &key} |
| 711 | if c := m.clientFor(bucket, nil); c != nil { |
| 712 | _, err := c.DeleteObject(ctx, in) |
| 713 | return err |
| 714 | } |
| 715 | _, err := m.dst.DeleteObject(ctx, in) |
| 716 | if err != nil && m.src != m.dst { |
| 717 | if _, err2 := m.src.DeleteObject(ctx, in); err2 == nil { |
| 718 | return nil |
| 719 | } |
| 720 | } |
| 721 | return err |
| 722 | } |
| 723 | |
| 724 | // Get and Put serve the receipts object, which lives in the source's own |
| 725 | // bucket: the bound client, else the source's. |
| 726 | func (m *s3Mirror) Get(ctx context.Context, bucket, key string) ([]byte, error) { |
| 727 | out, err := m.clientFor(bucket, m.src).GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key}) |
| 728 | if err != nil { |
| 729 | var nsk *s3types.NoSuchKey |
| 730 | if errors.As(err, &nsk) { |
| 731 | return nil, ErrMirrorNotFound |
| 732 | } |
| 733 | return nil, err |
| 734 | } |
| 735 | defer out.Body.Close() |
| 736 | return io.ReadAll(out.Body) |
| 737 | } |
| 738 | |
| 739 | func (m *s3Mirror) Put(ctx context.Context, bucket, key string, body []byte) error { |
| 740 | ct := "application/json" |
| 741 | _, err := m.clientFor(bucket, m.src).PutObject(ctx, &awss3.PutObjectInput{ |
| 742 | Bucket: &bucket, Key: &key, Body: bytes.NewReader(body), ContentType: &ct, |
| 743 | }) |
| 744 | return err |
| 745 | } |
| 746 | |
amadan