| 1 | package moves |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "os" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/lease" |
| 11 | "github.com/carlosframework/platform/internal/provision" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | // The readiness checks of §7.3. The target box writes T1 to T13 (T13 the cut |
| 16 | // guard's, Paul's ruling of 2026-10-11) into its |
| 17 | // report; the source box writes S1 and S2 into its own. Ready joins them. |
| 18 | const ( |
| 19 | CheckProtocol = "T1" |
| 20 | CheckSecrets = "T2" |
| 21 | CheckEnv = "T3" |
| 22 | CheckStorage = "T4" |
| 23 | CheckArchive = "T5" |
| 24 | CheckDisk = "T6" |
| 25 | CheckRelease = "T7" |
| 26 | CheckOneOwner = "T8" |
| 27 | CheckClaim = "T9" |
| 28 | CheckLane = "T10" |
| 29 | CheckBudget = "T11" |
| 30 | CheckQuiet = "T12" |
| 31 | // CheckArmed is the cut guard's T13 (Paul's ruling, 2026-10-11): the |
| 32 | // target box can land the move and roll it back, every hook of both |
| 33 | // wired, judged when the report is written. It is the report's arming |
| 34 | // statement: the source refuses to start a cut unless the target's |
| 35 | // latest report states it passing (TargetReport.Armed). |
| 36 | CheckArmed = "T13" |
| 37 | ) |
| 38 | |
| 39 | // AllChecks lists the target's checks in id order. Passing walks it, so the |
| 40 | // first failure it names is the lowest-numbered one. |
| 41 | var AllChecks = []string{CheckProtocol, CheckSecrets, CheckEnv, CheckStorage, CheckArchive, CheckDisk, CheckRelease, CheckOneOwner, CheckClaim, CheckLane, CheckBudget, CheckQuiet, CheckArmed} |
| 42 | |
| 43 | // ReportFreshness bounds how old a report may be and still decide ready |
| 44 | // (§7.3); the cut re-checks it at `admitted` (§7.4). |
| 45 | const ReportFreshness = 10 * time.Minute |
| 46 | |
| 47 | // ReportSkew is how far ahead of the reader's clock a report may be dated |
| 48 | // (clock skew between boxes). A report dated further ahead is refused, so a |
| 49 | // wrong or stuck clock cannot keep a report fresh. It is the same bound as |
| 50 | // MeasurementSkew. |
| 51 | const ReportSkew = time.Minute |
| 52 | |
| 53 | // MeasurementFreshness bounds how old a rate measurement may be and still |
| 54 | // price a cut (§7.3 T11). MeasurementSkew is how far ahead of the reader's |
| 55 | // clock a measurement made on another box may be dated. |
| 56 | const ( |
| 57 | MeasurementFreshness = time.Hour |
| 58 | MeasurementSkew = ReportSkew |
| 59 | ) |
| 60 | |
| 61 | // TargetReport is what the target's box publishes at TargetReportKey, every |
| 62 | // pass (§7.3). Inputs are what its checks were made against; Ready compares |
| 63 | // them with what the source publishes now. Notes are advisories that do not |
| 64 | // fail a check (the `shadowed` names of T3). |
| 65 | type TargetReport struct { |
| 66 | V int `json:"v"` |
| 67 | Box string `json:"box"` |
| 68 | Region string `json:"region"` |
| 69 | MoveID string `json:"move_id"` |
| 70 | At time.Time `json:"at"` |
| 71 | Checks map[string]Check `json:"checks"` |
| 72 | Inputs provision.MoveInputs `json:"inputs"` |
| 73 | AttemptsAdmitted int `json:"attempts_admitted"` |
| 74 | CutBudgetS int `json:"cut_budget_s"` |
| 75 | Rates Rates `json:"rates"` |
| 76 | // DBMiB, S3At and DiskAt are the evidence behind Rates, so that Ready |
| 77 | // can re-price the cut and re-check freshness when it reads the report, |
| 78 | // not only when the report was written. Rates.Cross is the source's rate |
| 79 | // as the target used it; Ready prices from the source's current one. |
| 80 | DBMiB float64 `json:"db_mib"` |
| 81 | S3At time.Time `json:"s3_at"` |
| 82 | DiskAt time.Time `json:"disk_at"` |
| 83 | // Entries is the archive's entry count the Archive phase was priced |
| 84 | // for, and FsyncAt when the per-file sync rate (Rates.FsyncRate) was |
| 85 | // measured (ruling T7-budget). |
| 86 | Entries int `json:"entries"` |
| 87 | FsyncAt time.Time `json:"fsync_at"` |
| 88 | Notes []string `json:"notes,omitempty"` |
| 89 | } |
| 90 | |
| 91 | // RecordBudget copies a BudgetCheck result and the measurements, database |
| 92 | // size and entry count it was priced from into the report, the fields Ready |
| 93 | // re-checks. |
| 94 | func (r *TargetReport) RecordBudget(dbMiB float64, entries int, m Measurements, res BudgetResult) { |
| 95 | r.DBMiB = dbMiB |
| 96 | r.Entries = entries |
| 97 | r.S3At = m.S3.At |
| 98 | r.DiskAt = m.Disk.At |
| 99 | r.FsyncAt = m.Fsync.At |
| 100 | r.Rates = res.Rates |
| 101 | r.AttemptsAdmitted = res.AttemptsAdmitted |
| 102 | } |
| 103 | |
| 104 | // SourceReport is what the source's box publishes at SourceReportKey. |
| 105 | // Release is the source's live version (release.LiveVersion), the pinned |
| 106 | // release T7 stages. CrossRate is the cross-region copy rate in MiB/s from |
| 107 | // the source's most recent copying round, made through the target bucket's |
| 108 | // region (the path the cut uses), and CrossRateAt is when that round ended. |
| 109 | type SourceReport struct { |
| 110 | V int `json:"v"` |
| 111 | Box string `json:"box"` |
| 112 | MoveID string `json:"move_id"` |
| 113 | At time.Time `json:"at"` |
| 114 | Env EnvReport `json:"env"` |
| 115 | S1 Check `json:"s1"` |
| 116 | S2 Check `json:"s2"` |
| 117 | FilesPlain string `json:"files_plain"` |
| 118 | Release string `json:"release"` |
| 119 | // DBBytes is the size of the source's local database file when one is |
| 120 | // present (a hibernating source may have none). T11 prices the larger |
| 121 | // of it and the logical size in the newest staged LTX header. |
| 122 | DBBytes int64 `json:"db_bytes,omitempty"` |
| 123 | // ReplicaBytes is the database's logical size as the source's replica |
| 124 | // states it this round: Commit pages of PageSize in the header of the |
| 125 | // newest LTX file under the claim's generation, the one the lane |
| 126 | // mirrors. Ready refuses while either size exceeds the size the |
| 127 | // target's report priced (TargetReport.DBMiB, final review finding 2). |
| 128 | ReplicaBytes int64 `json:"replica_bytes,omitempty"` |
| 129 | Archive *ArchiveRef `json:"archive,omitempty"` |
| 130 | CrossRate float64 `json:"cross_rate"` |
| 131 | CrossRateAt time.Time `json:"cross_rate_at"` |
| 132 | } |
| 133 | |
| 134 | // ArchiveRef names the archive the source published, by exact version, so |
| 135 | // T5 reads the very objects the source wrote. SealedObjectKey is the sealed |
| 136 | // archive's object key. It is not archive.Manifest's SealedDataKey, which is |
| 137 | // the key that seals the data inside. |
| 138 | type ArchiveRef struct { |
| 139 | ManifestKey string `json:"manifest_key"` |
| 140 | ManifestVersion string `json:"manifest_version"` |
| 141 | SealedObjectKey string `json:"sealed_object_key"` |
| 142 | SealedVersion string `json:"sealed_version"` |
| 143 | Plain string `json:"plain"` |
| 144 | } |
| 145 | |
| 146 | // Pinned reports whether both objects are named by a version id that pins |
| 147 | // one immutable version (lease.UsableVersionID): not "", and not the "null" |
| 148 | // a versioning-suspended bucket answers, which the next write to the key |
| 149 | // overwrites (final review finding 4). A reference that is not pinned is |
| 150 | // never verified or trusted from a cache. |
| 151 | func (r ArchiveRef) Pinned() bool { |
| 152 | return lease.UsableVersionID(r.ManifestVersion) && lease.UsableVersionID(r.SealedVersion) |
| 153 | } |
| 154 | |
| 155 | // CrossMeasurement is the source's cross-region rate as a measurement. A |
| 156 | // nil report, or one that has measured nothing, gives the zero measurement, |
| 157 | // which Measurements.Rates refuses. |
| 158 | func (s *SourceReport) CrossMeasurement() MeasuredRate { |
| 159 | if s == nil { |
| 160 | return MeasuredRate{} |
| 161 | } |
| 162 | return MeasuredRate{MiBPerS: s.CrossRate, At: s.CrossRateAt} |
| 163 | } |
| 164 | |
| 165 | // Armed reports whether the report states that its box can land the move |
| 166 | // and roll it back (CheckArmed). It fails closed: no report, a report from |
| 167 | // an older box that has no T13, and a failing T13 are all unarmed. |
| 168 | func (r *TargetReport) Armed() bool { |
| 169 | if r == nil { |
| 170 | return false |
| 171 | } |
| 172 | c, ok := r.Checks[CheckArmed] |
| 173 | return ok && c.OK |
| 174 | } |
| 175 | |
| 176 | // Passing reports whether every check passed, naming the first that did not |
| 177 | // in id order. A check that is missing from the report fails. |
| 178 | func (r TargetReport) Passing() (bool, string) { |
| 179 | for _, id := range AllChecks { |
| 180 | c, ok := r.Checks[id] |
| 181 | if !ok { |
| 182 | return false, id + " was not reported" |
| 183 | } |
| 184 | if !c.OK { |
| 185 | if c.Reason == "" { |
| 186 | return false, id + ": did not pass" |
| 187 | } |
| 188 | return false, id + ": " + c.Reason |
| 189 | } |
| 190 | } |
| 191 | return true, "" |
| 192 | } |
| 193 | |
| 194 | // Ready is §7.3's promotion rule: a round at most the bar, a passing target |
| 195 | // report and source readiness, each under ReportFreshness old, and at least |
| 196 | // one probe attempt admitted by the budget (R3-7). It is a pure reading and |
| 197 | // only ever gates promotion. A later stale report reads false, and nothing |
| 198 | // here, or in MovePromoteReady, which is legal from preparing only, takes a |
| 199 | // move back out of ready (Review Focus 5). |
| 200 | // |
| 201 | // The target's checks must also have been made against what the source |
| 202 | // publishes now (round-5 finding 9), its release included: a fresh passing report for archive A |
| 203 | // must not promote after the source has published archive B. And the budget |
| 204 | // must have been priced for a final mirror round as large as the bar allows |
| 205 | // (F28(e)), so raising CARLOS_MOVE_READY_DELTA_BYTES cannot promote on an |
| 206 | // estimate made under a smaller one. Nor can a database that has grown past |
| 207 | // the size the target priced (sizeCovered, final review finding 2). |
| 208 | func Ready(roundDelta, bar int64, target *TargetReport, source *SourceReport, now time.Time) (bool, string) { |
| 209 | if roundDelta > bar { |
| 210 | return false, fmt.Sprintf("the last round moved %d bytes and the bar is %d", roundDelta, bar) |
| 211 | } |
| 212 | if target == nil { |
| 213 | return false, "the target has not published a readiness report yet" |
| 214 | } |
| 215 | if age := now.Sub(target.At); age > ReportFreshness { |
| 216 | return false, "the target's readiness report is stale" |
| 217 | } else if age < -ReportSkew { |
| 218 | return false, "the target's readiness report is dated ahead of this clock" |
| 219 | } |
| 220 | if ok, why := target.Passing(); !ok { |
| 221 | return false, why |
| 222 | } |
| 223 | if target.AttemptsAdmitted < 1 { |
| 224 | return false, "T11: the cut budget admits no probe attempt" |
| 225 | } |
| 226 | priced := target.Rates.MirrorBar |
| 227 | if priced == 0 { |
| 228 | priced = MirrorBarMiB |
| 229 | } |
| 230 | if want := BarMiB(bar); !finite(priced) || !(priced >= want) { // written so NaN fails |
| 231 | return false, fmt.Sprintf("T11: the cut budget was priced for a final round of %g MiB and the bar is %g MiB", priced, want) |
| 232 | } |
| 233 | if source == nil { |
| 234 | return false, "the source has not published a readiness report yet" |
| 235 | } |
| 236 | if age := now.Sub(source.At); age > ReportFreshness { |
| 237 | return false, "the source's readiness report is stale" |
| 238 | } else if age < -ReportSkew { |
| 239 | return false, "the source's readiness report is dated ahead of this clock" |
| 240 | } |
| 241 | if !source.S1.OK { |
| 242 | return false, "S1: " + reasonOr(source.S1, "the source did not answer") |
| 243 | } |
| 244 | if !source.S2.OK { |
| 245 | return false, "S2: " + reasonOr(source.S2, "the source could not confirm its spawn state") |
| 246 | } |
| 247 | if why := budgetEvidence(target, source, bar, now); why != "" { |
| 248 | return false, why |
| 249 | } |
| 250 | if why := sizeCovered(target, source); why != "" { |
| 251 | return false, why |
| 252 | } |
| 253 | if target.Inputs.FilesPlain != source.FilesPlain { |
| 254 | return false, fmt.Sprintf("the target checked files %s and the source now publishes %s", target.Inputs.FilesPlain, source.FilesPlain) |
| 255 | } |
| 256 | if target.Inputs.EnvReport != source.Env.Digest { |
| 257 | return false, fmt.Sprintf("the target checked launch environment %s and the source now reports %s", target.Inputs.EnvReport, source.Env.Digest) |
| 258 | } |
| 259 | // The release promotion pins is the one T7 established on the target, |
| 260 | // never one the source moved to since (Task 14 review finding 3). |
| 261 | if target.Inputs.Release == "" { |
| 262 | return false, "the target has not checked a release yet" |
| 263 | } |
| 264 | if target.Inputs.Release != source.Release { |
| 265 | return false, fmt.Sprintf("the target checked release %s and the source now runs %s", target.Inputs.Release, source.Release) |
| 266 | } |
| 267 | return true, "" |
| 268 | } |
| 269 | |
| 270 | // budgetEvidence re-prices the cut from the evidence the reports carry, at |
| 271 | // `now`, and returns why it cannot be trusted, or "". Every rate must be a |
| 272 | // positive finite number measured within MeasurementFreshness (the target's |
| 273 | // S3 and disk rates by their report times, the cross-region rate by the |
| 274 | // source's CrossRateAt), the cut window and database size must be usable, |
| 275 | // and the recomputed budget must admit an attempt. The stored |
| 276 | // AttemptsAdmitted may be lower than the recomputation (the reports are |
| 277 | // refreshed on their own schedules) but never higher: a report cannot claim |
| 278 | // more attempts than its own evidence buys. |
| 279 | func budgetEvidence(target *TargetReport, source *SourceReport, bar int64, now time.Time) string { |
| 280 | if !usable(target.Rates.Cross) || !usable(target.Rates.S3) || !usable(target.Rates.Disk) || !usable(target.Rates.FsyncRate) { |
| 281 | return "T11: the target's report carries a rate that is not a usable number" |
| 282 | } |
| 283 | m := Measurements{ |
| 284 | Cross: source.CrossMeasurement(), |
| 285 | S3: MeasuredRate{MiBPerS: target.Rates.S3, At: target.S3At}, |
| 286 | Disk: MeasuredRate{MiBPerS: target.Rates.Disk, At: target.DiskAt}, |
| 287 | Fsync: FilesRate{PerS: target.Rates.FsyncRate, At: target.FsyncAt}, |
| 288 | } |
| 289 | res := BudgetCheck(target.DBMiB, target.Entries, m, bar, float64(target.CutBudgetS), now) |
| 290 | if !res.Check.OK { |
| 291 | return "T11: " + res.Check.Reason |
| 292 | } |
| 293 | if target.AttemptsAdmitted > res.AttemptsAdmitted { |
| 294 | return fmt.Sprintf("T11: the target's report claims %d probe attempts and its measurements admit %d", target.AttemptsAdmitted, res.AttemptsAdmitted) |
| 295 | } |
| 296 | return "" |
| 297 | } |
| 298 | |
| 299 | // sizeCovered refuses a database that has grown past the size the target's |
| 300 | // report priced (final review finding 2). TargetReport.DBMiB is the size |
| 301 | // both T11's budget and T6's disk check were made for, and the source's |
| 302 | // report carries the size now: its local file (DBBytes) and its replica's |
| 303 | // logical size (ReplicaBytes). A larger one leaves both checks stale even |
| 304 | // while the report is fresh, so promotion waits for the target's next |
| 305 | // report, which prices the new size. Equal or smaller is covered. |
| 306 | func sizeCovered(target *TargetReport, source *SourceReport) string { |
| 307 | size := max(source.DBBytes, source.ReplicaBytes) |
| 308 | if now := float64(size) / (1 << 20); now > target.DBMiB { |
| 309 | return fmt.Sprintf("T11: the database has grown to %g MiB since the target priced %g MiB. waiting for the target's next report", now, target.DBMiB) |
| 310 | } |
| 311 | return "" |
| 312 | } |
| 313 | |
| 314 | func reasonOr(c Check, fallback string) string { |
| 315 | if c.Reason == "" { |
| 316 | return fallback |
| 317 | } |
| 318 | return c.Reason |
| 319 | } |
| 320 | |
| 321 | // ---- measured rates and T11 ---- |
| 322 | |
| 323 | // MeasuredRate is one throughput measurement, in MiB/s, with when and over |
| 324 | // how many bytes it was taken. |
| 325 | type MeasuredRate struct { |
| 326 | MiBPerS float64 `json:"mib_per_s"` |
| 327 | At time.Time `json:"at"` |
| 328 | Bytes int64 `json:"bytes"` |
| 329 | } |
| 330 | |
| 331 | // RateOf is bytes over d as a measurement made at `at`. No bytes or no |
| 332 | // elapsed time is no measurement (a zero rate), never an infinite one. |
| 333 | func RateOf(bytes int64, d time.Duration, at time.Time) MeasuredRate { |
| 334 | m := MeasuredRate{At: at, Bytes: bytes} |
| 335 | if bytes > 0 && d > 0 { |
| 336 | m.MiBPerS = float64(bytes) / (1 << 20) / d.Seconds() |
| 337 | } |
| 338 | return m |
| 339 | } |
| 340 | |
| 341 | // FilesRate is a per-file sync measurement: files synced per second, with |
| 342 | // when and over how many files it was taken (ruling T7-budget). |
| 343 | type FilesRate struct { |
| 344 | PerS float64 `json:"per_s"` |
| 345 | At time.Time `json:"at"` |
| 346 | Files int `json:"files"` |
| 347 | } |
| 348 | |
| 349 | // Measurements are the rates T11 prices a cut from. The target measures |
| 350 | // regional S3, its disk and its per-file sync rate itself. Cross comes from |
| 351 | // the source's report (SourceReport.CrossMeasurement), as only the source's |
| 352 | // copying rounds cross the regions. |
| 353 | type Measurements struct { |
| 354 | Cross, S3, Disk MeasuredRate |
| 355 | Fsync FilesRate |
| 356 | } |
| 357 | |
| 358 | // BarMiB is a ready-delta bar in bytes as the MiB that Rates.MirrorBar |
| 359 | // counts. A bar of zero or less is zero, which Estimate prices at the |
| 360 | // default, never below it. |
| 361 | func BarMiB(bytes int64) float64 { |
| 362 | if bytes <= 0 { |
| 363 | return 0 |
| 364 | } |
| 365 | return float64(bytes) / (1 << 20) |
| 366 | } |
| 367 | |
| 368 | // Rates turns the measurements into the rates Estimate takes, with the |
| 369 | // operator's ready-delta bar (bytes) as the mirror bar. Every rate must be |
| 370 | // a positive finite number taken in the last MeasurementFreshness; any other |
| 371 | // is refused, naming which. The worked example's rates are never a fallback. |
| 372 | func (m Measurements) Rates(now time.Time, barBytes int64) (Rates, error) { |
| 373 | for _, c := range []struct { |
| 374 | which string |
| 375 | m MeasuredRate |
| 376 | }{ |
| 377 | {"the cross-region copy path", m.Cross}, |
| 378 | {"regional S3", m.S3}, |
| 379 | {"the disk", m.Disk}, |
| 380 | } { |
| 381 | if !fresh(c.m.MiBPerS, c.m.At, now) { |
| 382 | return Rates{}, &NoFreshRateError{Which: c.which} |
| 383 | } |
| 384 | } |
| 385 | if !fresh(m.Fsync.PerS, m.Fsync.At, now) { |
| 386 | return Rates{}, &NoFreshRateError{Which: "the disk's per-file sync"} |
| 387 | } |
| 388 | return Rates{Cross: m.Cross.MiBPerS, S3: m.S3.MiBPerS, Disk: m.Disk.MiBPerS, MirrorBar: BarMiB(barBytes), FsyncRate: m.Fsync.PerS}, nil |
| 389 | } |
| 390 | |
| 391 | // NoFreshRateError is Measurements.Rates refusing a rate: Which names it. |
| 392 | type NoFreshRateError struct{ Which string } |
| 393 | |
| 394 | func (e *NoFreshRateError) Error() string { return NoFreshRate(e.Which) } |
| 395 | |
| 396 | // NoFreshRate is T11's sentence for a missing or stale rate (MV-N68). |
| 397 | func NoFreshRate(which string) string { |
| 398 | return fmt.Sprintf("no fresh rate measurement for %s", which) |
| 399 | } |
| 400 | |
| 401 | // fresh is a usable rate measured within MeasurementFreshness of now and no |
| 402 | // more than MeasurementSkew ahead of it. |
| 403 | func fresh(rate float64, at, now time.Time) bool { |
| 404 | age := now.Sub(at) |
| 405 | return !at.IsZero() && usable(rate) && age <= MeasurementFreshness && age >= -MeasurementSkew |
| 406 | } |
| 407 | |
| 408 | // BudgetResult is T11 decided: the check, the attempts the window admits, |
| 409 | // the rates and the price, for the report to carry. |
| 410 | type BudgetResult struct { |
| 411 | Check Check |
| 412 | AttemptsAdmitted int |
| 413 | Rates Rates |
| 414 | Seconds float64 // the cut priced with one probe attempt |
| 415 | } |
| 416 | |
| 417 | // BudgetCheck is T11 (§7.3): price a cut of a dbMiB database whose archive |
| 418 | // holds entries files at the measured rates, with barBytes as the final |
| 419 | // mirror bar, and admit the probe attempts that fit windowSeconds. It fails |
| 420 | // closed: a missing, stale or unusable measurement, a negative entry count, |
| 421 | // a database size or window that is not a usable number, or a window that |
| 422 | // admits no attempt, fails the check with zero attempts admitted. |
| 423 | func BudgetCheck(dbMiB float64, entries int, m Measurements, barBytes int64, windowSeconds float64, now time.Time) BudgetResult { |
| 424 | fail := func(why string) BudgetResult { return BudgetResult{Check: Check{Reason: why}} } |
| 425 | if !finite(dbMiB) || dbMiB < 0 { |
| 426 | return fail("the database size could not be measured") |
| 427 | } |
| 428 | if entries < 0 { |
| 429 | return fail("the archive's file count could not be measured") |
| 430 | } |
| 431 | if !finite(windowSeconds) || windowSeconds <= 0 { |
| 432 | return fail("the cut window is not a usable number of seconds") |
| 433 | } |
| 434 | rates, err := m.Rates(now, barBytes) |
| 435 | var nf *NoFreshRateError |
| 436 | if errors.As(err, &nf) { |
| 437 | return fail(NoFreshRate(nf.Which)) |
| 438 | } |
| 439 | if err != nil { |
| 440 | return fail("the cut budget could not be priced") |
| 441 | } |
| 442 | e := Estimate(dbMiB, entries, rates) |
| 443 | n := e.AttemptsAdmitted(windowSeconds) |
| 444 | need := e.Total(1) |
| 445 | res := BudgetResult{AttemptsAdmitted: n, Rates: rates, Seconds: need} |
| 446 | if n < 1 { |
| 447 | res.Check.Reason = fmt.Sprintf("the cut needs about %.0f seconds with one probe attempt and the window is %.0f seconds", need, windowSeconds) |
| 448 | return res |
| 449 | } |
| 450 | res.Check.OK = true |
| 451 | return res |
| 452 | } |
| 453 | |
| 454 | // ---- writing a recurring object ---- |
| 455 | |
| 456 | // PutReplace writes key whether or not it exists. release.Store.Put with an |
| 457 | // empty ifMatch is create-only, so a recurring object (a report, the |
| 458 | // console record) is written by Get for the current etag (absent is "") and |
| 459 | // then Put with it, retrying once when another writer got in between. A Get |
| 460 | // that fails for any reason but absence is returned, never written over. |
| 461 | // Every mutable object the series writes to the deployment bucket goes |
| 462 | // through here. |
| 463 | // |
| 464 | // It has twins: putCAS in internal/edge/certsync.go (three attempts) and |
| 465 | // status.WriteBox in internal/status/snapshot.go (one retry, a second |
| 466 | // conflict is returned loudly). Both are the same read-etag-then-If-Match |
| 467 | // loop with different policies and error wording. Neither fits here: |
| 468 | // putCAS is unexported in internal/edge, and WriteBox writes a box's own |
| 469 | // status snapshot to its fixed key, not an arbitrary key and body. |
| 470 | // This one retries ONCE on conflict, like WriteBox: a report has a single |
| 471 | // writer per pass, so two conflicts in a row mean something else is writing |
| 472 | // the key, and the caller sees ErrCASConflict rather than a silent third try. |
| 473 | func PutReplace(ctx context.Context, s release.Store, key string, body []byte) error { |
| 474 | var err error |
| 475 | for attempt := 0; attempt < 2; attempt++ { |
| 476 | _, etag, gerr := s.Get(ctx, key) |
| 477 | switch { |
| 478 | case gerr == nil: |
| 479 | case errors.Is(gerr, os.ErrNotExist): |
| 480 | etag = "" |
| 481 | default: |
| 482 | return fmt.Errorf("read %s before replacing it: %w", key, gerr) |
| 483 | } |
| 484 | if _, err = s.Put(ctx, key, body, etag); err == nil { |
| 485 | return nil |
| 486 | } |
| 487 | if !errors.Is(err, release.ErrCASConflict) { |
| 488 | break |
| 489 | } |
| 490 | } |
| 491 | return fmt.Errorf("write %s: %w", key, err) |
| 492 | } |
| 493 | |
amadan