| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "sync" |
| 12 | "time" |
| 13 | ) |
| 14 | |
| 15 | const pendingHotBytes = 16 << 20 |
| 16 | |
| 17 | type queueReservation struct { |
| 18 | binding *PersistenceBinding |
| 19 | bytes int64 |
| 20 | mu sync.Mutex |
| 21 | state uint8 // 0 reserved, 1 consumed, 2 released |
| 22 | } |
| 23 | |
| 24 | func (r *queueReservation) release() { |
| 25 | if r == nil || r.binding == nil { |
| 26 | return |
| 27 | } |
| 28 | r.mu.Lock() |
| 29 | if r.state != 0 { |
| 30 | r.mu.Unlock() |
| 31 | return |
| 32 | } |
| 33 | r.state = 2 |
| 34 | r.mu.Unlock() |
| 35 | r.binding.mu.Lock() |
| 36 | r.binding.reservedBytes -= r.bytes |
| 37 | r.binding.notifySpaceLocked() |
| 38 | r.binding.mu.Unlock() |
| 39 | } |
| 40 | |
| 41 | // PersistenceBinding delivers accepted commits to one physical SessionHandle. |
| 42 | // |
| 43 | // It holds only the write-behind prefix of the same event sequence that Session |
| 44 | // owns; it is not a second source of business state. Delivery into the queue is |
| 45 | // a pure memory operation, so a slow disk can never extend the session commit |
| 46 | // lock or block cancellation. |
| 47 | type PersistenceBinding struct { |
| 48 | // Immutable after construction: Session readers retain this reference while |
| 49 | // Close shuts down the handle. Write admission is guarded by closed/accepting. |
| 50 | handle SessionHandle |
| 51 | dir string |
| 52 | |
| 53 | // metadataSource rebuilds the list projection for the catalog cache once |
| 54 | // the accepted prefix is fully durable. Session supplies it so the binding |
| 55 | // never holds a projection of its own. |
| 56 | metadataSource func(durable uint64) (catalogMetadata, bool) |
| 57 | recoverySource func(durable uint64) (recoveryPublishState, bool) |
| 58 | recoveryPublished func(durable uint64) |
| 59 | disableRecoveryPublish bool |
| 60 | |
| 61 | mu sync.Mutex |
| 62 | queue []Commit |
| 63 | durable uint64 |
| 64 | timer timerHandle |
| 65 | draining bool |
| 66 | autoPaused bool |
| 67 | accepting bool |
| 68 | closed bool |
| 69 | writeErr error |
| 70 | uncertain *uncertainWrite |
| 71 | hotBytes int64 |
| 72 | reservedBytes int64 |
| 73 | spaceChanged chan struct{} |
| 74 | |
| 75 | drainMu sync.Mutex |
| 76 | closeOnce sync.Once |
| 77 | closeErr error |
| 78 | |
| 79 | afterFunc func(time.Duration, func()) timerHandle |
| 80 | writeFn func(context.Context, io.Writer, []byte) error |
| 81 | syncFn func(*os.File) error |
| 82 | } |
| 83 | |
| 84 | func newPersistenceBinding(handle SessionHandle, dir string, durable uint64, opts OpenOptions) *PersistenceBinding { |
| 85 | after := opts.AfterFunc |
| 86 | if after == nil { |
| 87 | after = func(delay time.Duration, fire func()) timerHandle { return time.AfterFunc(delay, fire) } |
| 88 | } |
| 89 | writeFn := opts.Write |
| 90 | if writeFn == nil { |
| 91 | writeFn = writeAllContext |
| 92 | } |
| 93 | syncFn := opts.Sync |
| 94 | if syncFn == nil { |
| 95 | syncFn = func(file *os.File) error { return file.Sync() } |
| 96 | } |
| 97 | return &PersistenceBinding{ |
| 98 | handle: handle, dir: dir, durable: durable, accepting: true, |
| 99 | afterFunc: after, writeFn: writeFn, syncFn: syncFn, |
| 100 | spaceChanged: make(chan struct{}), |
| 101 | } |
| 102 | } |
| 103 | |
| 104 | func (b *PersistenceBinding) reserve(ctx context.Context, bytes int64) (*queueReservation, error) { |
| 105 | if b == nil { |
| 106 | return nil, osClosedError() |
| 107 | } |
| 108 | charge := min(max(bytes, 1), int64(pendingHotBytes)) |
| 109 | for { |
| 110 | b.mu.Lock() |
| 111 | if !b.accepting || b.closed { |
| 112 | b.mu.Unlock() |
| 113 | return nil, osClosedError() |
| 114 | } |
| 115 | if b.hotBytes+b.reservedBytes+charge <= pendingHotBytes { |
| 116 | b.reservedBytes += charge |
| 117 | reservation := &queueReservation{binding: b, bytes: charge} |
| 118 | b.mu.Unlock() |
| 119 | return reservation, nil |
| 120 | } |
| 121 | wait := b.spaceChanged |
| 122 | b.mu.Unlock() |
| 123 | select { |
| 124 | case <-ctx.Done(): |
| 125 | return nil, ctx.Err() |
| 126 | case <-wait: |
| 127 | } |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | func (b *PersistenceBinding) notifySpaceLocked() { |
| 132 | if b.spaceChanged == nil { |
| 133 | b.spaceChanged = make(chan struct{}) |
| 134 | return |
| 135 | } |
| 136 | close(b.spaceChanged) |
| 137 | b.spaceChanged = make(chan struct{}) |
| 138 | } |
| 139 | |
| 140 | // accept atomically adds an immutable commit to the write-behind queue and |
| 141 | // publishes it to Session through accepted. It performs no file I/O and calls |
| 142 | // accepted exactly once while the queue is locked. Keeping these two memory |
| 143 | // mutations in one boundary prevents a closed binding from rejecting a commit |
| 144 | // after Session has already exposed it, and prevents Flush from persisting a |
| 145 | // commit before Session exposes it. |
| 146 | func (b *PersistenceBinding) accept(commit Commit, reservation *queueReservation, accepted func()) error { |
| 147 | if b == nil { |
| 148 | return osClosedError() |
| 149 | } |
| 150 | b.mu.Lock() |
| 151 | defer b.mu.Unlock() |
| 152 | if !b.accepting || b.closed { |
| 153 | if reservation != nil { |
| 154 | reservation.mu.Lock() |
| 155 | if reservation.state == 0 { |
| 156 | reservation.state = 2 |
| 157 | b.reservedBytes -= reservation.bytes |
| 158 | b.notifySpaceLocked() |
| 159 | } |
| 160 | reservation.mu.Unlock() |
| 161 | } |
| 162 | return osClosedError() |
| 163 | } |
| 164 | if reservation == nil || reservation.binding != b { |
| 165 | return errors.New("session: missing or foreign queue reservation") |
| 166 | } |
| 167 | reservation.mu.Lock() |
| 168 | if reservation.state != 0 { |
| 169 | reservation.mu.Unlock() |
| 170 | return errors.New("session: queue reservation is no longer valid") |
| 171 | } |
| 172 | reservation.state = 1 |
| 173 | reservation.mu.Unlock() |
| 174 | b.reservedBytes -= reservation.bytes |
| 175 | b.hotBytes += reservation.bytes |
| 176 | // Session transfers an immutable commit here. Keep one shared payload backing |
| 177 | // for the accepted log and write queue; cloning large bodies would turn the |
| 178 | // queue budget into multiple hidden copies. |
| 179 | b.queue = append(b.queue, commit) |
| 180 | accepted() |
| 181 | if !b.autoPaused && !b.draining && b.timer == nil { |
| 182 | b.scheduleDrainLocked() |
| 183 | } |
| 184 | return nil |
| 185 | } |
| 186 | |
| 187 | func commitHotBytes(commit Commit) int64 { |
| 188 | var bytes int64 = 512 |
| 189 | for _, event := range commit.Events { |
| 190 | bytes += int64(len(event.ID)+len(event.Kind)+len(event.Payload)) + 256 |
| 191 | if event.PayloadRef != nil { |
| 192 | bytes += int64(len(event.PayloadRef.Digest)+len(event.PayloadRef.IndexDigest)+len(event.PayloadRef.MediaType)+len(event.PayloadRef.Name)) + 64 |
| 193 | } |
| 194 | } |
| 195 | return min(max(bytes, 1), int64(pendingHotBytes)) |
| 196 | } |
| 197 | |
| 198 | func (b *PersistenceBinding) stopAccepting() { |
| 199 | if b == nil { |
| 200 | return |
| 201 | } |
| 202 | b.mu.Lock() |
| 203 | b.accepting = false |
| 204 | b.notifySpaceLocked() |
| 205 | b.mu.Unlock() |
| 206 | } |
| 207 | |
| 208 | func (b *PersistenceBinding) scheduleDrainLocked() { |
| 209 | b.timer = b.afterFunc(LiveBatchDelay, func() { _ = b.drain(context.Background(), false) }) |
| 210 | } |
| 211 | |
| 212 | // progress reports the durable watermark and whether the accepted prefix is |
| 213 | // fully persisted. |
| 214 | func (b *PersistenceBinding) progress() (uint64, PersistenceStatus, string) { |
| 215 | if b == nil { |
| 216 | return 0, PersistenceReady, "" |
| 217 | } |
| 218 | b.mu.Lock() |
| 219 | defer b.mu.Unlock() |
| 220 | status := PersistenceReady |
| 221 | if b.writeErr != nil { |
| 222 | status = PersistenceFailed |
| 223 | if errors.Is(b.writeErr, ErrPersistenceUncertain) { |
| 224 | status = PersistenceUncertain |
| 225 | } |
| 226 | } else if len(b.queue) > 0 || b.draining { |
| 227 | status = PersistencePending |
| 228 | } |
| 229 | return b.durable, status, errorString(b.writeErr) |
| 230 | } |
| 231 | |
| 232 | // Flush drains the queue and reports the durable sequence. Cancelling one |
| 233 | // caller's wait never cancels the shared physical write. |
| 234 | func (b *PersistenceBinding) Flush(ctx context.Context) (DurableReceipt, error) { |
| 235 | if b == nil { |
| 236 | return DurableReceipt{}, fmt.Errorf("session: nil persistence binding") |
| 237 | } |
| 238 | if err := ctx.Err(); err != nil { |
| 239 | return DurableReceipt{}, err |
| 240 | } |
| 241 | // Once an append starts, one caller cannot cancel the shared physical write. |
| 242 | // drainMu merges callers onto the ordered write chain while each caller can |
| 243 | // still stop waiting through its own context. |
| 244 | done := make(chan error, 1) |
| 245 | go func() { done <- b.drain(context.Background(), true) }() |
| 246 | select { |
| 247 | case err := <-done: |
| 248 | return DurableReceipt{DurableSequence: b.durableSequence()}, err |
| 249 | case <-ctx.Done(): |
| 250 | return DurableReceipt{DurableSequence: b.durableSequence()}, ctx.Err() |
| 251 | } |
| 252 | } |
| 253 | |
| 254 | // FlushThrough waits for one accepted commit boundary. Later writes may keep |
| 255 | // draining, but cannot extend this caller's snapshot or cancel its shared writer. |
| 256 | func (b *PersistenceBinding) FlushThrough(ctx context.Context, through uint64) (DurableReceipt, error) { |
| 257 | if b == nil { |
| 258 | return DurableReceipt{}, os.ErrClosed |
| 259 | } |
| 260 | if err := ctx.Err(); err != nil { |
| 261 | return DurableReceipt{}, err |
| 262 | } |
| 263 | b.mu.Lock() |
| 264 | durable, changed := b.durable, b.spaceChanged |
| 265 | b.mu.Unlock() |
| 266 | if durable >= through { |
| 267 | return DurableReceipt{DurableSequence: durable}, nil |
| 268 | } |
| 269 | done := make(chan error, 1) |
| 270 | go func() { done <- b.drain(context.Background(), true) }() |
| 271 | for { |
| 272 | select { |
| 273 | case <-ctx.Done(): |
| 274 | return DurableReceipt{DurableSequence: b.durableSequence()}, ctx.Err() |
| 275 | case err := <-done: |
| 276 | durable = b.durableSequence() |
| 277 | if durable >= through { |
| 278 | return DurableReceipt{DurableSequence: durable}, nil |
| 279 | } |
| 280 | if err == nil { |
| 281 | err = fmt.Errorf("session: watermark %d was not accepted", through) |
| 282 | } |
| 283 | return DurableReceipt{DurableSequence: durable}, err |
| 284 | case <-changed: |
| 285 | b.mu.Lock() |
| 286 | durable, changed = b.durable, b.spaceChanged |
| 287 | b.mu.Unlock() |
| 288 | if durable >= through { |
| 289 | return DurableReceipt{DurableSequence: durable}, nil |
| 290 | } |
| 291 | } |
| 292 | } |
| 293 | } |
| 294 | |
| 295 | func (b *PersistenceBinding) durableSequence() uint64 { |
| 296 | b.mu.Lock() |
| 297 | defer b.mu.Unlock() |
| 298 | return b.durable |
| 299 | } |
| 300 | |
| 301 | func (b *PersistenceBinding) drain(ctx context.Context, explicit bool) error { |
| 302 | b.drainMu.Lock() |
| 303 | defer b.drainMu.Unlock() |
| 304 | for { |
| 305 | b.mu.Lock() |
| 306 | if b.closed || b.handle == nil { |
| 307 | b.mu.Unlock() |
| 308 | return os.ErrClosed |
| 309 | } |
| 310 | if b.timer != nil { |
| 311 | b.timer.Stop() |
| 312 | b.timer = nil |
| 313 | } |
| 314 | if len(b.queue) == 0 { |
| 315 | b.draining = false |
| 316 | b.mu.Unlock() |
| 317 | b.refreshCatalogMetadata() |
| 318 | return nil |
| 319 | } |
| 320 | uncertain := cloneUncertainWrite(b.uncertain) |
| 321 | if uncertain != nil && !explicit { |
| 322 | err := b.writeErr |
| 323 | b.draining = false |
| 324 | b.mu.Unlock() |
| 325 | return err |
| 326 | } |
| 327 | b.draining = true |
| 328 | // Detach only the queue header. Commit payloads are immutable after |
| 329 | // acceptance and can be streamed by the physical writer without copying. |
| 330 | pending := append([]Commit(nil), b.queue...) |
| 331 | handle := b.handle |
| 332 | b.mu.Unlock() |
| 333 | |
| 334 | alreadyPersisted := false |
| 335 | var err error |
| 336 | if uncertain != nil { |
| 337 | alreadyPersisted, err = b.reconcileUncertain(ctx, handle, *uncertain) |
| 338 | } |
| 339 | if err == nil && alreadyPersisted { |
| 340 | confirmed := uncertain.commitCount |
| 341 | if confirmed <= 0 || confirmed > len(pending) { |
| 342 | err = fmt.Errorf("%w: uncertain batch commit count %d exceeds pending prefix %d", ErrDamagedStore, confirmed, len(pending)) |
| 343 | } else { |
| 344 | b.mu.Lock() |
| 345 | if len(b.queue) < confirmed || !sameCommitPrefix(b.queue, pending[:confirmed]) { |
| 346 | err = fmt.Errorf("%w: uncertain batch no longer matches pending prefix", ErrDamagedStore) |
| 347 | b.writeErr = err |
| 348 | b.autoPaused = true |
| 349 | b.draining = false |
| 350 | b.mu.Unlock() |
| 351 | return err |
| 352 | } |
| 353 | for _, commit := range b.queue[:confirmed] { |
| 354 | b.hotBytes -= commitHotBytes(commit) |
| 355 | } |
| 356 | b.queue = b.queue[confirmed:] |
| 357 | b.notifySpaceLocked() |
| 358 | b.durable = pending[confirmed-1].LastSequence() |
| 359 | b.writeErr = nil |
| 360 | b.uncertain = nil |
| 361 | b.autoPaused = false |
| 362 | if len(b.queue) == 0 { |
| 363 | b.draining = false |
| 364 | b.mu.Unlock() |
| 365 | b.refreshCatalogMetadata() |
| 366 | return nil |
| 367 | } |
| 368 | b.mu.Unlock() |
| 369 | continue |
| 370 | } |
| 371 | } |
| 372 | if err == nil && !alreadyPersisted { |
| 373 | err = b.persist(ctx, handle, pending) |
| 374 | } |
| 375 | b.mu.Lock() |
| 376 | if err != nil { |
| 377 | b.draining = false |
| 378 | b.writeErr = err |
| 379 | var uncertainErr *uncertainAppendError |
| 380 | if errors.As(err, &uncertainErr) { |
| 381 | copy := uncertainErr.write |
| 382 | b.uncertain = © |
| 383 | } |
| 384 | b.autoPaused = true |
| 385 | b.mu.Unlock() |
| 386 | return err |
| 387 | } |
| 388 | if len(b.queue) < len(pending) || !sameCommitPrefix(b.queue, pending) { |
| 389 | b.draining = false |
| 390 | b.writeErr = fmt.Errorf("%w: pending batch order changed", ErrDamagedStore) |
| 391 | b.autoPaused = true |
| 392 | err := b.writeErr |
| 393 | b.mu.Unlock() |
| 394 | return err |
| 395 | } |
| 396 | for _, commit := range b.queue[:len(pending)] { |
| 397 | b.hotBytes -= commitHotBytes(commit) |
| 398 | } |
| 399 | b.queue = b.queue[len(pending):] |
| 400 | b.notifySpaceLocked() |
| 401 | b.durable = pending[len(pending)-1].LastSequence() |
| 402 | b.writeErr = nil |
| 403 | b.uncertain = nil |
| 404 | b.autoPaused = false |
| 405 | if len(b.queue) == 0 { |
| 406 | b.draining = false |
| 407 | b.mu.Unlock() |
| 408 | b.refreshCatalogMetadata() |
| 409 | return nil |
| 410 | } |
| 411 | // Match DSH's drain chain: once a batch starts writing, events accepted |
| 412 | // during that write are drained immediately in the next physical batch. |
| 413 | // The fixed 200ms window applies only to the first pending batch. |
| 414 | b.mu.Unlock() |
| 415 | } |
| 416 | } |
| 417 | |
| 418 | func (b *PersistenceBinding) refreshCatalogMetadata() { |
| 419 | if b == nil { |
| 420 | return |
| 421 | } |
| 422 | b.mu.Lock() |
| 423 | if b.closed || len(b.queue) != 0 { |
| 424 | b.mu.Unlock() |
| 425 | return |
| 426 | } |
| 427 | durable := b.durable |
| 428 | b.mu.Unlock() |
| 429 | recoveryPublished := false |
| 430 | if !b.disableRecoveryPublish && b.recoverySource != nil { |
| 431 | if state, ok := b.recoverySource(durable); ok { |
| 432 | if store, ok := b.handle.(*Store); ok && store.recovery != nil { |
| 433 | if err := store.recovery.publish(context.Background(), state.checkpoint, state.operations); err == nil { |
| 434 | recoveryPublished = true |
| 435 | } |
| 436 | } |
| 437 | } |
| 438 | } |
| 439 | if b.metadataSource != nil { |
| 440 | if metadata, ok := b.metadataSource(durable); ok { |
| 441 | _ = writeCatalogMetadataForSession(filepath.Join(filepath.Dir(b.dir), ".query-cache", filepath.Base(b.dir)), b.dir, metadata) |
| 442 | } |
| 443 | } |
| 444 | if (recoveryPublished || b.disableRecoveryPublish || b.recoverySource == nil) && b.recoveryPublished != nil { |
| 445 | b.recoveryPublished(durable) |
| 446 | } |
| 447 | } |
| 448 | |
| 449 | func (b *PersistenceBinding) persist(ctx context.Context, handle SessionHandle, commits []Commit) error { |
| 450 | return handle.Append(ctx, commits) |
| 451 | } |
| 452 | |
| 453 | // reconcileUncertain proves whether the prior append completed before retrying. |
| 454 | // The writer lease and drainMu make this a single-owner repair operation. A |
| 455 | // partial tail is preserved byte-for-byte before truncation; a mismatching or |
| 456 | // unexpectedly extended tail remains recovery-required rather than guessed. |
| 457 | func (b *PersistenceBinding) reconcileUncertain(ctx context.Context, handle SessionHandle, uncertain uncertainWrite) (bool, error) { |
| 458 | if err := ctx.Err(); err != nil { |
| 459 | return false, err |
| 460 | } |
| 461 | physical, ok := handle.(*Store) |
| 462 | if !ok || physical == nil { |
| 463 | return false, fmt.Errorf("%w: handle cannot reconcile an uncertain append", ErrPersistenceUncertain) |
| 464 | } |
| 465 | file, err := physical.writableFile() |
| 466 | if err != nil { |
| 467 | return false, err |
| 468 | } |
| 469 | info, err := file.Stat() |
| 470 | if err != nil { |
| 471 | return false, err |
| 472 | } |
| 473 | if info.Size() < uncertain.start { |
| 474 | return false, fmt.Errorf("%w: log shrank below uncertain offset %d", ErrPersistenceUncertain, uncertain.start) |
| 475 | } |
| 476 | tailLen := info.Size() - uncertain.start |
| 477 | if tailLen == 0 { |
| 478 | return false, nil |
| 479 | } |
| 480 | staged, err := os.Open(uncertain.stagedPath) |
| 481 | if err != nil { |
| 482 | return false, fmt.Errorf("%w: open staged append evidence: %w", ErrPersistenceUncertain, err) |
| 483 | } |
| 484 | defer staged.Close() |
| 485 | stagedInfo, err := staged.Stat() |
| 486 | if err != nil || stagedInfo.Size() != uncertain.stagedBytes { |
| 487 | return false, fmt.Errorf("%w: staged append evidence changed: %w", ErrPersistenceUncertain, err) |
| 488 | } |
| 489 | if tailLen > uncertain.stagedBytes { |
| 490 | return false, fmt.Errorf("%w: on-disk tail exceeds staged batch at offset %d", ErrPersistenceUncertain, uncertain.start) |
| 491 | } |
| 492 | matches, err := equalReaderPrefix(ctx, io.NewSectionReader(file, uncertain.start, tailLen), staged, tailLen) |
| 493 | if err != nil { |
| 494 | return false, fmt.Errorf("%w: inspect uncertain tail: %w", ErrPersistenceUncertain, err) |
| 495 | } |
| 496 | if tailLen == uncertain.stagedBytes && matches { |
| 497 | if err := b.syncFn(file); err != nil { |
| 498 | return false, &uncertainAppendError{cause: fmt.Errorf("fsync verified append: %w", err), write: uncertain} |
| 499 | } |
| 500 | // The uncertain fsync did not advance the physical index. Rebuild its |
| 501 | // durable cursor before validating a queued successor, or the successor |
| 502 | // would appear to start after a gap. |
| 503 | if err := physical.rebuildWriterIndex(file); err != nil { |
| 504 | return false, fmt.Errorf("%w: rebuild index after verified append: %w", ErrPersistenceUncertain, err) |
| 505 | } |
| 506 | _ = staged.Close() |
| 507 | _ = os.Remove(uncertain.stagedPath) |
| 508 | return true, nil |
| 509 | } |
| 510 | if tailLen < uncertain.stagedBytes && matches { |
| 511 | backup, err := preserveAndTruncateTail(filepath.Join(b.dir, currentLogName), uncertain.start, "uncertain") |
| 512 | if err != nil { |
| 513 | return false, fmt.Errorf("%w: preserve partial tail: %w", ErrPersistenceUncertain, err) |
| 514 | } |
| 515 | _ = backup |
| 516 | if _, err := file.Seek(0, io.SeekEnd); err != nil { |
| 517 | return false, fmt.Errorf("%w: seek repaired log: %w", ErrPersistenceUncertain, err) |
| 518 | } |
| 519 | _ = staged.Close() |
| 520 | _ = os.Remove(uncertain.stagedPath) |
| 521 | return false, nil |
| 522 | } |
| 523 | return false, fmt.Errorf("%w: on-disk tail does not match batch at offset %d", ErrPersistenceUncertain, uncertain.start) |
| 524 | } |
| 525 | |
| 526 | func equalReaderPrefix(ctx context.Context, left, right io.Reader, length int64) (bool, error) { |
| 527 | leftBuffer := make([]byte, 1<<20) |
| 528 | rightBuffer := make([]byte, 1<<20) |
| 529 | remaining := length |
| 530 | for remaining > 0 { |
| 531 | if err := ctx.Err(); err != nil { |
| 532 | return false, err |
| 533 | } |
| 534 | chunk := min(remaining, int64(len(leftBuffer))) |
| 535 | ln, leftErr := io.ReadFull(left, leftBuffer[:chunk]) |
| 536 | rn, rightErr := io.ReadFull(right, rightBuffer[:chunk]) |
| 537 | if leftErr != nil || rightErr != nil { |
| 538 | return false, errors.Join(leftErr, rightErr) |
| 539 | } |
| 540 | if ln != rn || !bytes.Equal(leftBuffer[:ln], rightBuffer[:rn]) { |
| 541 | return false, nil |
| 542 | } |
| 543 | remaining -= int64(ln) |
| 544 | } |
| 545 | return true, nil |
| 546 | } |
| 547 | |
| 548 | // freezePhysical runs fn while the physical write chain is idle. Export uses it |
| 549 | // so the copied bytes cannot advance between the manifest and the log. |
| 550 | func (b *PersistenceBinding) freezePhysical(fn func() error) error { |
| 551 | if b == nil || fn == nil { |
| 552 | return nil |
| 553 | } |
| 554 | b.drainMu.Lock() |
| 555 | defer b.drainMu.Unlock() |
| 556 | return fn() |
| 557 | } |
| 558 | |
| 559 | // Close drains the queue and closes the physical handle. It is deliberately |
| 560 | // uncancellable and idempotent: every caller observes the same result. |
| 561 | func (b *PersistenceBinding) Close(ctx context.Context) error { |
| 562 | if b == nil { |
| 563 | return nil |
| 564 | } |
| 565 | b.closeOnce.Do(func() { |
| 566 | b.mu.Lock() |
| 567 | b.accepting = false |
| 568 | needsFlush := len(b.queue) > 0 || b.uncertain != nil |
| 569 | b.notifySpaceLocked() |
| 570 | b.mu.Unlock() |
| 571 | var flushErr error |
| 572 | if needsFlush { |
| 573 | _, flushErr = b.Flush(context.Background()) |
| 574 | } |
| 575 | b.drainMu.Lock() |
| 576 | defer b.drainMu.Unlock() |
| 577 | b.mu.Lock() |
| 578 | if b.timer != nil { |
| 579 | b.timer.Stop() |
| 580 | b.timer = nil |
| 581 | } |
| 582 | handle := b.handle |
| 583 | b.closed = true |
| 584 | b.notifySpaceLocked() |
| 585 | b.mu.Unlock() |
| 586 | var closeErr error |
| 587 | if handle != nil { |
| 588 | closeErr = handle.Close(context.Background()) |
| 589 | } |
| 590 | b.closeErr = errors.Join(flushErr, closeErr) |
| 591 | }) |
| 592 | return b.closeErr |
| 593 | } |
| 594 | |
| 595 | func cloneUncertainWrite(in *uncertainWrite) *uncertainWrite { |
| 596 | if in == nil { |
| 597 | return nil |
| 598 | } |
| 599 | out := *in |
| 600 | return &out |
| 601 | } |
| 602 |