| 1 | package assetcache |
| 2 | |
| 3 | import ( |
| 4 | "compress/gzip" |
| 5 | "crypto/rand" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "io" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "runtime" |
| 13 | "sync" |
| 14 | ) |
| 15 | |
| 16 | const ( |
| 17 | // MaxBody is the largest identity body the store will hold. |
| 18 | MaxBody = 32 << 20 |
| 19 | |
| 20 | // metaReserve is the disk a reservation sets aside for the .meta file. |
| 21 | metaReserve = 64 << 10 |
| 22 | |
| 23 | // MaxMeta is the largest encoded .meta file Commit writes. |
| 24 | MaxMeta = metaReserve |
| 25 | |
| 26 | // maxFillsPerBox and maxFillsPerApp bound the fills in flight. |
| 27 | maxFillsPerBox = 16 |
| 28 | maxFillsPerApp = 4 |
| 29 | |
| 30 | // floorPercent is the share of the filesystem that must stay free. |
| 31 | floorPercent = 15 |
| 32 | |
| 33 | // sampleTries is how many free-space samples admit takes without the |
| 34 | // store's lock before it takes one with the lock held. |
| 35 | sampleTries = 4 |
| 36 | ) |
| 37 | |
| 38 | var ( |
| 39 | // ErrNoRoom means the budget has no room for the fill, even after eviction. |
| 40 | ErrNoRoom = errors.New("assetcache: no room in the budget") |
| 41 | // ErrBelowFloor means admitting the fill could take the disk below its |
| 42 | // free-space floor. |
| 43 | ErrBelowFloor = errors.New("assetcache: free space would fall below the floor") |
| 44 | // ErrBusy means the box or the app already has as many fills in flight |
| 45 | // as it may. |
| 46 | ErrBusy = errors.New("assetcache: too many fills in flight") |
| 47 | // ErrTooLarge means the body is, or has grown, past what the fill reserved. |
| 48 | ErrTooLarge = errors.New("assetcache: body too large") |
| 49 | ) |
| 50 | |
| 51 | // reservation is the disk a fill has set aside: room for the identity body, |
| 52 | // for its gzip copy, and for the metadata file. Commit holds each to its own |
| 53 | // half. |
| 54 | type reservation struct { |
| 55 | identity int64 |
| 56 | gzip int64 |
| 57 | meta int64 |
| 58 | } |
| 59 | |
| 60 | func (r reservation) total() int64 { return r.identity + r.gzip + r.meta } |
| 61 | |
| 62 | // Fill is one fill in flight. Write the identity body into it. It holds a |
| 63 | // reservation and one fill slot until it is committed or aborted. |
| 64 | type Fill struct { |
| 65 | s *Store |
| 66 | key Key |
| 67 | scope string |
| 68 | id string |
| 69 | res reservation |
| 70 | |
| 71 | mu sync.Mutex |
| 72 | tmp *os.File // identity body temporary |
| 73 | gz *os.File // gzip copy temporary, while Commit writes it |
| 74 | bodyPath string // where the identity body is: its temporary, then its versioned name |
| 75 | gzPath string // likewise for the gzip copy |
| 76 | token string // names every temporary of this fill |
| 77 | written int64 |
| 78 | failed error // set by the first failed write |
| 79 | done bool // committed or aborted: the fill slot is released |
| 80 | held int64 // bytes of the reservation still counted in s.reserved |
| 81 | left []leftover // temporaries an Abort could not remove yet |
| 82 | |
| 83 | crashAt string // tests only: the Commit stage at which to stop as a crash would |
| 84 | failAt string // tests only: the Commit stage whose I/O fails with errInjected |
| 85 | } |
| 86 | |
| 87 | // leftover is a temporary still on disk and the bytes it holds. |
| 88 | type leftover struct { |
| 89 | path string |
| 90 | size int64 |
| 91 | } |
| 92 | |
| 93 | // Reserve admits a fill for k, or says why not. contentLength is the length |
| 94 | // the origin announced; negative means unknown. It sets aside twice the body |
| 95 | // (identity plus gzip copy) and metaReserve, taking the body as MaxBody when |
| 96 | // the length is unknown. |
| 97 | // |
| 98 | // A reservation that eviction cannot help is refused before anything is |
| 99 | // evicted for it: one larger than the budget, one that does not fit beside |
| 100 | // the other fills' reservations and the retired bytes no eviction frees, and |
| 101 | // one the free-space floor refuses. Only a reservation that the budget alone |
| 102 | // stands in the way of evicts, and then every condition is checked again on |
| 103 | // a fresh free-space sample. Before it refuses for want of room, Reserve |
| 104 | // retries the unlinks that failed earlier, whose bytes are still counted. |
| 105 | func (s *Store) Reserve(k Key, contentLength int64) (*Fill, error) { |
| 106 | if contentLength > MaxBody { |
| 107 | return nil, ErrTooLarge |
| 108 | } |
| 109 | n := contentLength |
| 110 | if n < 0 { |
| 111 | n = MaxBody |
| 112 | } |
| 113 | res := reservation{identity: n, gzip: n, meta: metaReserve} |
| 114 | scope := k.AppScope() |
| 115 | r := res.total() |
| 116 | |
| 117 | s.mu.Lock() |
| 118 | err := s.busyLocked(scope) |
| 119 | s.mu.Unlock() |
| 120 | if err != nil { |
| 121 | return nil, err |
| 122 | } |
| 123 | |
| 124 | retried, evicted := false, false |
| 125 | for { |
| 126 | err := s.admit(scope, r) |
| 127 | if err == nil { |
| 128 | break |
| 129 | } |
| 130 | if !errors.Is(err, ErrNoRoom) && !errors.Is(err, errShort) { |
| 131 | return nil, err |
| 132 | } |
| 133 | if !retried { |
| 134 | retried = true |
| 135 | if s.retryStuck() { |
| 136 | continue |
| 137 | } |
| 138 | } |
| 139 | if errors.Is(err, ErrNoRoom) || evicted { |
| 140 | return nil, ErrNoRoom |
| 141 | } |
| 142 | s.makeRoom(r) |
| 143 | evicted = true |
| 144 | } |
| 145 | f := &Fill{s: s, key: k, scope: scope, id: k.ID(), res: res, held: r} |
| 146 | if err := f.open(); err != nil { |
| 147 | f.finish() |
| 148 | return nil, err |
| 149 | } |
| 150 | return f, nil |
| 151 | } |
| 152 | |
| 153 | // busyLocked reports ErrBusy when a new fill for scope would pass a bound. |
| 154 | func (s *Store) busyLocked(scope string) error { |
| 155 | if s.fills >= maxFillsPerBox || s.fillsByApp[scope] >= maxFillsPerApp { |
| 156 | return ErrBusy |
| 157 | } |
| 158 | return nil |
| 159 | } |
| 160 | |
| 161 | // errShort is admit's answer when only the budget stands in the way, and |
| 162 | // evicting entries can make the room. |
| 163 | var errShort = errors.New("assetcache: no room in the budget until entries are evicted") |
| 164 | |
| 165 | // admit takes a free-space sample and decides on it whether a reservation of |
| 166 | // r bytes for scope may start. If it may, admit claims the reservation and a |
| 167 | // fill slot and returns nil; otherwise it returns the refusal, or errShort. |
| 168 | // |
| 169 | // The sample and the reservation counter must agree. A fill that commits |
| 170 | // between the sample and the decision releases its reservation, and the |
| 171 | // bytes it wrote after the sample are then in neither, so a decision that |
| 172 | // combined the two would count them nowhere. Every release ticks |
| 173 | // s.releases, and admit samples again when the count moved; after |
| 174 | // sampleTries unlocked samples it samples with s.mu held, where nothing can |
| 175 | // release. A reservation claimed meanwhile only raises the counter, and a |
| 176 | // fill still in flight writes inside a reservation the counter still holds, |
| 177 | // so neither calls for a new sample. |
| 178 | func (s *Store) admit(scope string, r int64) error { |
| 179 | for try := 1; ; try++ { |
| 180 | locked := try > sampleTries |
| 181 | s.mu.Lock() |
| 182 | releases := s.releases |
| 183 | if !locked { |
| 184 | s.mu.Unlock() |
| 185 | } |
| 186 | total, free, err := s.cfg.FS.Space(s.cfg.Dir) |
| 187 | if !locked { |
| 188 | if s.afterSample != nil { |
| 189 | s.afterSample() |
| 190 | } |
| 191 | s.mu.Lock() |
| 192 | } |
| 193 | if err != nil { |
| 194 | s.mu.Unlock() |
| 195 | return fmt.Errorf("assetcache: free space: %w", err) |
| 196 | } |
| 197 | if s.releases != releases { |
| 198 | s.mu.Unlock() |
| 199 | continue |
| 200 | } |
| 201 | err = s.decideLocked(scope, r, total, free) |
| 202 | s.mu.Unlock() |
| 203 | return err |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | // decideLocked applies the admission rules to one free-space sample, total |
| 208 | // and free, in an order that keeps a reservation eviction cannot help from |
| 209 | // evicting anything. If every rule holds it claims the reservation and a |
| 210 | // fill slot. The caller holds s.mu. |
| 211 | func (s *Store) decideLocked(scope string, r int64, total, free uint64) error { |
| 212 | if err := s.busyLocked(scope); err != nil { |
| 213 | return err |
| 214 | } |
| 215 | // The room eviction can make at most: the budget less the other fills' |
| 216 | // reservations and the retired bytes (held by Readers, by unlinks that |
| 217 | // failed, or kept for the next Open), none of which eviction frees. |
| 218 | if r > s.cfg.Budget || s.retired > s.cfg.Budget-s.reserved-r { |
| 219 | return ErrNoRoom |
| 220 | } |
| 221 | // Free space after every outstanding reservation and this one must stay |
| 222 | // at or above the floor. Reservations may already be partly written, so |
| 223 | // this counts some bytes twice: it errs towards refusing. The floor is |
| 224 | // judged on the disk as sampled: eviction makes room in the budget, so a |
| 225 | // fill the floor refuses evicts nothing. |
| 226 | if need := uint64(s.reserved) + uint64(r) + total*floorPercent/100; free < need { |
| 227 | return ErrBelowFloor |
| 228 | } |
| 229 | if s.used > s.cfg.Budget-s.reserved-r { // used + reserved + r > Budget, without overflow |
| 230 | return errShort |
| 231 | } |
| 232 | s.reserved += r |
| 233 | s.fills++ |
| 234 | s.fillsByApp[scope]++ |
| 235 | return nil |
| 236 | } |
| 237 | |
| 238 | // open creates the identity temporary. Its name carries ".tmp-", so a crash |
| 239 | // leaves something Open removes. |
| 240 | func (f *Fill) open() error { |
| 241 | var b [8]byte |
| 242 | if _, err := rand.Read(b[:]); err != nil { |
| 243 | return fmt.Errorf("assetcache: %w", err) |
| 244 | } |
| 245 | f.token = hex.EncodeToString(b[:]) |
| 246 | f.bodyPath = f.tempName("body") |
| 247 | f.gzPath = f.tempName("gz") |
| 248 | // Read and write: Commit reads the body back to compress it. |
| 249 | tmp, err := os.OpenFile(f.bodyPath, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0o600) |
| 250 | if err != nil { |
| 251 | return fmt.Errorf("assetcache: %w", err) |
| 252 | } |
| 253 | f.tmp = tmp |
| 254 | return nil |
| 255 | } |
| 256 | |
| 257 | // tempName is the path of this fill's temporary for part ("body", "gz" or |
| 258 | // "meta"). |
| 259 | func (f *Fill) tempName(part string) string { |
| 260 | return filepath.Join(f.s.cfg.Dir, f.id+".tmp-"+f.token+"."+part) |
| 261 | } |
| 262 | |
| 263 | // Write appends to the identity body. A write that would pass the |
| 264 | // reservation's identity half, or MaxBody, writes nothing, returns |
| 265 | // ErrTooLarge and fails the fill for good. |
| 266 | func (f *Fill) Write(p []byte) (int, error) { |
| 267 | f.mu.Lock() |
| 268 | defer f.mu.Unlock() |
| 269 | if f.done { |
| 270 | return 0, os.ErrClosed |
| 271 | } |
| 272 | if f.failed != nil { |
| 273 | return 0, f.failed |
| 274 | } |
| 275 | if int64(len(p)) > f.res.identity-f.written || int64(len(p)) > MaxBody-f.written { |
| 276 | f.failed = ErrTooLarge |
| 277 | return 0, f.failed |
| 278 | } |
| 279 | n, err := f.tmp.Write(p) |
| 280 | f.written += int64(n) |
| 281 | if err != nil { |
| 282 | f.failed = fmt.Errorf("assetcache: %w", err) |
| 283 | return n, f.failed |
| 284 | } |
| 285 | return n, nil |
| 286 | } |
| 287 | |
| 288 | // Abort deletes the fill's temporaries and releases its reservation and fill |
| 289 | // slot. It is safe to call twice and after Commit. If a temporary cannot be |
| 290 | // removed, the fill slot is still released but the bytes of the files that |
| 291 | // remain stay counted as reserved, so the budget and the floor see them; a |
| 292 | // later Abort retries the removal, and Open removes them at the next start. |
| 293 | func (f *Fill) Abort() { |
| 294 | f.mu.Lock() |
| 295 | defer f.mu.Unlock() |
| 296 | f.finish() |
| 297 | } |
| 298 | |
| 299 | // finish closes the fill and removes its temporaries. The first call |
| 300 | // releases the fill slot. Reserved bytes are released once the files they |
| 301 | // cover are gone: if a removal fails, the bytes of the files that remain stay |
| 302 | // reserved and a later call retries just those. The caller holds f.mu, or has |
| 303 | // not shared f. |
| 304 | func (f *Fill) finish() { |
| 305 | if !f.done { |
| 306 | f.done = true |
| 307 | f.closeFiles() |
| 308 | if f.token != "" { |
| 309 | f.left = []leftover{ |
| 310 | {path: f.bodyPath, size: f.res.identity}, |
| 311 | {path: f.gzPath, size: f.res.gzip}, |
| 312 | {path: f.tempName("meta"), size: f.res.meta}, |
| 313 | } |
| 314 | } |
| 315 | f.s.releaseSlot(f.scope) |
| 316 | } |
| 317 | var remaining int64 |
| 318 | kept := f.left[:0] |
| 319 | for _, l := range f.left { |
| 320 | fi, err := os.Lstat(l.path) |
| 321 | if errors.Is(err, os.ErrNotExist) { |
| 322 | continue |
| 323 | } |
| 324 | if err == nil { |
| 325 | l.size = fi.Size() // what the file really holds |
| 326 | } |
| 327 | if err := os.Remove(l.path); err == nil || errors.Is(err, os.ErrNotExist) { |
| 328 | continue |
| 329 | } |
| 330 | kept = append(kept, l) |
| 331 | remaining += l.size |
| 332 | } |
| 333 | f.left = kept |
| 334 | f.s.releaseReserved(f.held - remaining) |
| 335 | f.held = remaining |
| 336 | } |
| 337 | |
| 338 | // closeFiles closes the fill's open temporaries. The caller holds f.mu. |
| 339 | func (f *Fill) closeFiles() { |
| 340 | if f.tmp != nil { |
| 341 | f.tmp.Close() |
| 342 | f.tmp = nil |
| 343 | } |
| 344 | if f.gz != nil { |
| 345 | f.gz.Close() |
| 346 | f.gz = nil |
| 347 | } |
| 348 | } |
| 349 | |
| 350 | // closeBodies closes the body temporaries before they are published and |
| 351 | // reports the first failure, so a body the kernel could not finish writing |
| 352 | // is never renamed into place. The caller holds f.mu. |
| 353 | func (f *Fill) closeBodies() error { |
| 354 | var err error |
| 355 | if f.tmp != nil { |
| 356 | err = f.tmp.Close() |
| 357 | f.tmp = nil |
| 358 | } |
| 359 | if f.gz != nil { |
| 360 | if cerr := f.gz.Close(); err == nil { |
| 361 | err = cerr |
| 362 | } |
| 363 | f.gz = nil |
| 364 | } |
| 365 | if err != nil { |
| 366 | err = fmt.Errorf("assetcache: %w", err) |
| 367 | } |
| 368 | return f.fault(faultBodyClose, err) |
| 369 | } |
| 370 | |
| 371 | // releaseSlot returns a fill slot. |
| 372 | func (s *Store) releaseSlot(scope string) { |
| 373 | s.mu.Lock() |
| 374 | defer s.mu.Unlock() |
| 375 | s.releaseSlotLocked(scope) |
| 376 | } |
| 377 | |
| 378 | // releaseSlotLocked returns a fill slot. The caller holds s.mu. |
| 379 | func (s *Store) releaseSlotLocked(scope string) { |
| 380 | s.fills-- |
| 381 | if s.fillsByApp[scope]--; s.fillsByApp[scope] <= 0 { |
| 382 | delete(s.fillsByApp, scope) |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | // releaseReserved returns n reserved bytes. |
| 387 | func (s *Store) releaseReserved(n int64) { |
| 388 | if n == 0 { |
| 389 | return |
| 390 | } |
| 391 | s.mu.Lock() |
| 392 | defer s.mu.Unlock() |
| 393 | s.releaseReservedLocked(n) |
| 394 | } |
| 395 | |
| 396 | // releaseReservedLocked returns n reserved bytes and ticks s.releases, so an |
| 397 | // admission that sampled free space before this release samples again (see |
| 398 | // admit). The caller holds s.mu. |
| 399 | func (s *Store) releaseReservedLocked(n int64) { |
| 400 | s.reserved -= n |
| 401 | s.releases++ |
| 402 | } |
| 403 | |
| 404 | // crashBeforeMetaRename is the Commit stage, for tests, after the meta |
| 405 | // temporary is written and before it is renamed over <id>.meta. |
| 406 | const crashBeforeMetaRename = "before meta rename" |
| 407 | |
| 408 | // errSimulatedCrash is what Commit returns when it stops at f.crashAt. It |
| 409 | // leaves the files as a crash would and cleans up nothing. |
| 410 | var errSimulatedCrash = errors.New("assetcache: simulated crash") |
| 411 | |
| 412 | // Commit stages at which a test can inject an I/O failure through f.failAt. |
| 413 | const ( |
| 414 | faultBodyClose = "body close" |
| 415 | faultMetaDirSync = "meta dir sync" |
| 416 | ) |
| 417 | |
| 418 | // errInjected is the failure a test injects through f.failAt. |
| 419 | var errInjected = errors.New("assetcache: injected failure") |
| 420 | |
| 421 | // fault is err, or errInjected when a test has asked stage to fail. |
| 422 | func (f *Fill) fault(stage string, err error) error { |
| 423 | if err == nil && f.failAt == stage { |
| 424 | return errInjected |
| 425 | } |
| 426 | return err |
| 427 | } |
| 428 | |
| 429 | // errGzipTooLarge stops the gzip copy once it would pass its limit. |
| 430 | var errGzipTooLarge = errors.New("assetcache: gzip copy larger than its limit") |
| 431 | |
| 432 | // Commit stores what was written as the entry for the fill's key, replacing |
| 433 | // any entry it already has. It sets m's Key, Version, Size, GzipSize and |
| 434 | // StoredAt itself; the caller supplies Header, Lifetime and Expires. With |
| 435 | // gzipCopy it also keeps a gzip copy of the body, unless that would be larger |
| 436 | // than the body or the reservation's gzip half. |
| 437 | // |
| 438 | // The order makes a crash leave either the old entry or the new one, never |
| 439 | // new bytes under old metadata: (1) gzip the body into a temporary; |
| 440 | // (2) draw a fresh version; (3) fill in the meta; (4) encode it, refusing with |
| 441 | // ErrTooLarge when it passes MaxMeta, before anything is renamed; (5) fsync |
| 442 | // the bodies and rename them to their versioned names; (6) write, fsync and |
| 443 | // rename the meta over <id>.meta; (7) retire the previous version, whose |
| 444 | // files are unlinked at once but stay counted while a Reader holds them. The |
| 445 | // meta's bytes are then counted with the entry and the reservation released. |
| 446 | // |
| 447 | // Any error before the meta rename behaves as Abort. If the directory sync |
| 448 | // after the meta rename fails, Commit returns that error but the new entry |
| 449 | // stays published (see commit). Commit after Commit or Abort returns |
| 450 | // os.ErrClosed and changes nothing. |
| 451 | func (f *Fill) Commit(m Meta, gzipCopy bool) error { |
| 452 | f.mu.Lock() |
| 453 | defer f.mu.Unlock() |
| 454 | if f.done { |
| 455 | return os.ErrClosed |
| 456 | } |
| 457 | if f.failed != nil { |
| 458 | f.finish() |
| 459 | return f.failed |
| 460 | } |
| 461 | err := f.commit(m, gzipCopy) |
| 462 | if errors.Is(err, errSimulatedCrash) { |
| 463 | return err |
| 464 | } |
| 465 | if err != nil { |
| 466 | f.finish() |
| 467 | } |
| 468 | return err |
| 469 | } |
| 470 | |
| 471 | // commit does Commit's work. Until the meta rename every error returns with |
| 472 | // the fill's files where finish can find and remove them; after it, the fill |
| 473 | // is done whatever happens. The caller holds f.mu. |
| 474 | func (f *Fill) commit(m Meta, gzipCopy bool) error { |
| 475 | s := f.s |
| 476 | |
| 477 | // (1) The gzip copy, held to the identity size and the gzip half. |
| 478 | var gzSize int64 |
| 479 | if gzipCopy { |
| 480 | n, kept, err := f.writeGzip() |
| 481 | if err != nil { |
| 482 | return err |
| 483 | } |
| 484 | if kept { |
| 485 | gzSize = n |
| 486 | } |
| 487 | } |
| 488 | |
| 489 | // (2) A fresh version, so no file an older meta names is overwritten. |
| 490 | var b [8]byte |
| 491 | if _, err := rand.Read(b[:]); err != nil { |
| 492 | return fmt.Errorf("assetcache: %w", err) |
| 493 | } |
| 494 | version := hex.EncodeToString(b[:]) |
| 495 | |
| 496 | // (3) The fields the store owns. |
| 497 | m.Key = f.key |
| 498 | m.Header = m.Header.Clone() // the store's own: the caller may reuse its map |
| 499 | m.Version = version |
| 500 | m.Size = f.written |
| 501 | m.GzipSize = gzSize |
| 502 | m.StoredAt = s.cfg.Now() |
| 503 | |
| 504 | // (4) The final meta must fit its reservation before anything is renamed. |
| 505 | raw, err := encodeMeta(m) |
| 506 | if err != nil { |
| 507 | return fmt.Errorf("assetcache: %w", err) |
| 508 | } |
| 509 | if int64(len(raw)) > MaxMeta || int64(len(raw)) > f.res.meta { |
| 510 | return ErrTooLarge |
| 511 | } |
| 512 | |
| 513 | // (5) The bodies, durable under their versioned names. Nothing names |
| 514 | // them until the meta rename, so a crash here leaves orphans Open removes. |
| 515 | if err := f.tmp.Sync(); err != nil { |
| 516 | return fmt.Errorf("assetcache: %w", err) |
| 517 | } |
| 518 | if f.gz != nil { |
| 519 | if err := f.gz.Sync(); err != nil { |
| 520 | return fmt.Errorf("assetcache: %w", err) |
| 521 | } |
| 522 | } |
| 523 | if err := f.closeBodies(); err != nil { |
| 524 | return err |
| 525 | } |
| 526 | dir := s.cfg.Dir |
| 527 | if err := renameTo(&f.bodyPath, filepath.Join(dir, bodyName(f.id, version))); err != nil { |
| 528 | return err |
| 529 | } |
| 530 | if gzSize > 0 { |
| 531 | if err := renameTo(&f.gzPath, filepath.Join(dir, gzipName(f.id, version))); err != nil { |
| 532 | return err |
| 533 | } |
| 534 | } |
| 535 | if err := syncDir(dir); err != nil { |
| 536 | return err |
| 537 | } |
| 538 | |
| 539 | // (6) The meta, under the id lock so that the order of meta renames for |
| 540 | // one id is the order its index slot changes in. |
| 541 | unlock := s.lockID(f.id) |
| 542 | metaTmp := f.tempName("meta") |
| 543 | if err := writeSynced(metaTmp, raw); err != nil { |
| 544 | unlock() |
| 545 | return err |
| 546 | } |
| 547 | if f.crashAt == crashBeforeMetaRename { |
| 548 | unlock() |
| 549 | return errSimulatedCrash |
| 550 | } |
| 551 | if err := os.Rename(metaTmp, filepath.Join(dir, metaName(f.id))); err != nil { |
| 552 | unlock() |
| 553 | return fmt.Errorf("assetcache: %w", err) |
| 554 | } |
| 555 | // The new entry is published: its meta and bodies are on disk. If the |
| 556 | // directory sync fails, the rename may not be durable and a crash could |
| 557 | // bring back the old meta, so the old bodies must stay. The new entry is |
| 558 | // still indexed and served (it is what the directory holds), the old |
| 559 | // version is kept on disk and counted until the next Open (whose scan |
| 560 | // removes whichever bodies no meta names), and Commit returns the error. |
| 561 | syncErr := f.fault(faultMetaDirSync, syncDir(dir)) |
| 562 | |
| 563 | // (7) Swap the index slot, count the new entry and release the |
| 564 | // reservation and the fill slot in one step, then retire the old |
| 565 | // version. Its bytes stay counted until its files are gone. |
| 566 | ne := newEntry(f.id, m, int64(len(raw))) |
| 567 | s.mu.Lock() |
| 568 | old := s.entries[f.id] |
| 569 | if old != nil { |
| 570 | s.unindexLocked(old) |
| 571 | if syncErr != nil { |
| 572 | old.ver.unlinking = false |
| 573 | old.ver.kept = true |
| 574 | } |
| 575 | } |
| 576 | s.index(ne) |
| 577 | s.releaseReservedLocked(f.held) |
| 578 | s.releaseSlotLocked(f.scope) |
| 579 | s.mu.Unlock() |
| 580 | unlock() |
| 581 | f.held = 0 |
| 582 | f.left = nil |
| 583 | f.done = true |
| 584 | |
| 585 | if syncErr != nil { |
| 586 | return fmt.Errorf("assetcache: committed, but the meta rename may not be durable: %w", syncErr) |
| 587 | } |
| 588 | if old != nil { |
| 589 | s.unlink(old.ver) |
| 590 | } |
| 591 | s.retryStuck() |
| 592 | return nil |
| 593 | } |
| 594 | |
| 595 | // writeGzip compresses the identity body into the gzip temporary, which it |
| 596 | // leaves open in f.gz. It reports kept=false, with the temporary removed, |
| 597 | // when the copy would be larger than the identity body or the reservation's |
| 598 | // gzip half; no write ever passes that limit. The caller holds f.mu. |
| 599 | func (f *Fill) writeGzip() (n int64, kept bool, err error) { |
| 600 | out, err := os.OpenFile(f.gzPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) |
| 601 | if err != nil { |
| 602 | return 0, false, fmt.Errorf("assetcache: %w", err) |
| 603 | } |
| 604 | f.gz = out |
| 605 | lw := &limitWriter{w: out, left: min(f.written, f.res.gzip)} |
| 606 | zw := gzip.NewWriter(lw) |
| 607 | _, err = io.Copy(zw, io.NewSectionReader(f.tmp, 0, f.written)) |
| 608 | if err == nil { |
| 609 | err = zw.Close() |
| 610 | } |
| 611 | if errors.Is(err, errGzipTooLarge) { |
| 612 | f.gz.Close() |
| 613 | f.gz = nil |
| 614 | if err := os.Remove(f.gzPath); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 615 | return 0, false, fmt.Errorf("assetcache: %w", err) |
| 616 | } |
| 617 | return 0, false, nil |
| 618 | } |
| 619 | if err != nil { |
| 620 | return 0, false, fmt.Errorf("assetcache: gzip copy: %w", err) |
| 621 | } |
| 622 | return lw.n, true, nil |
| 623 | } |
| 624 | |
| 625 | // limitWriter passes writes through until one would take the total past |
| 626 | // left; that write writes nothing and returns errGzipTooLarge. |
| 627 | type limitWriter struct { |
| 628 | w io.Writer |
| 629 | left int64 |
| 630 | n int64 |
| 631 | } |
| 632 | |
| 633 | func (l *limitWriter) Write(p []byte) (int, error) { |
| 634 | if int64(len(p)) > l.left { |
| 635 | return 0, errGzipTooLarge |
| 636 | } |
| 637 | n, err := l.w.Write(p) |
| 638 | l.left -= int64(n) |
| 639 | l.n += int64(n) |
| 640 | return n, err |
| 641 | } |
| 642 | |
| 643 | // renameTo renames *from to to and, once it has, records the new name in |
| 644 | // *from, so an abort removes the file wherever it is. |
| 645 | func renameTo(from *string, to string) error { |
| 646 | if err := os.Rename(*from, to); err != nil { |
| 647 | return fmt.Errorf("assetcache: %w", err) |
| 648 | } |
| 649 | *from = to |
| 650 | return nil |
| 651 | } |
| 652 | |
| 653 | // writeSynced writes data to a new file at path and fsyncs it. |
| 654 | func writeSynced(path string, data []byte) error { |
| 655 | out, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600) |
| 656 | if err != nil { |
| 657 | return fmt.Errorf("assetcache: %w", err) |
| 658 | } |
| 659 | _, err = out.Write(data) |
| 660 | if err == nil { |
| 661 | err = out.Sync() |
| 662 | } |
| 663 | if cerr := out.Close(); err == nil { |
| 664 | err = cerr |
| 665 | } |
| 666 | if err != nil { |
| 667 | return fmt.Errorf("assetcache: %w", err) |
| 668 | } |
| 669 | return nil |
| 670 | } |
| 671 | |
| 672 | // syncDir fsyncs the directory, so the renames in it are durable. Windows |
| 673 | // cannot sync a directory handle; the store does not run there. |
| 674 | func syncDir(dir string) error { |
| 675 | if runtime.GOOS == "windows" { |
| 676 | return nil |
| 677 | } |
| 678 | d, err := os.Open(dir) |
| 679 | if err != nil { |
| 680 | return fmt.Errorf("assetcache: %w", err) |
| 681 | } |
| 682 | err = d.Sync() |
| 683 | if cerr := d.Close(); err == nil { |
| 684 | err = cerr |
| 685 | } |
| 686 | if err != nil { |
| 687 | return fmt.Errorf("assetcache: %w", err) |
| 688 | } |
| 689 | return nil |
| 690 | } |
| 691 | |
amadan