engine.go

package wal

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/binary"
	"errors"
	"math"
	"os"
	"sync"
	"sync/atomic"

	"github.com/mgurevin/wal/internal/lockrank"
)

type (
	fileLog struct {
		// Nested lock ranks are flush, maintenance, preparation/scan, queue,
		// state, object keys, object directory, then object state. A path may
		// enter at any rank; preparation and scan are peers and never nest.
		// See CONCURRENCY.md.
		mutex      *contextMutex // File state.
		flushMutex *contextMutex // Synced flush ordering.
		// queueMutex owns append admission and completion.
		queueMutex lockrank.Mutex
		// stagingMutex protects preparation admission and accounting.
		stagingMutex lockrank.Mutex
		// maintenanceMutex excludes flush, truncate, compact, and first Close.
		maintenanceMutex lockrank.Mutex
		// scanMutex protects scan admission and active lease count.
		scanMutex            lockrank.Mutex
		queueCondition       *sync.Cond
		metadata             *fileMetadata
		objects              *fileObjects
		activeFile           *os.File
		config               FileConfig
		appendBuffer         []byte
		scanTokens           chan struct{}
		stagingTokens        chan struct{}
		pending              []*appendRequest
		segments             []fileSegment
		entries              []fileIndexEntry
		fragments            []fileFragment
		ids                  map[string]fileRecordIdentity
		truncateCleanup      *truncateCleanup
		fragmentBase         uint64
		pendingRequests      int
		pendingBytes         int64
		preparingRequests    int
		preparingBytes       int64
		stagingReservedBytes atomic.Int64
		// Capacity snapshots run before commit admission, so these counters
		// cannot rely on the commit mutex for synchronization.
		segmentBytes            atomic.Int64
		compactionReservedBytes atomic.Int64
		admittedAppend          uint64
		completedAppend         uint64
		logicalBytes            int64
		indexBytes              int64
		last                    Position
		truncated               Position
		checksum                uint64
		syncedThrough           Position
		segmentCount            int
		dirtySegments           int
		activeBytes             int64
		activeRecords           int
		activeFileGeneration    uint64
		activeScans             int
		commitLeader            bool
		flushBarrier            bool
		admissionClosed         bool
		stagingClosed           bool
		scanAdmissionClosed     bool
		syncFailed              bool
		stateUncertain          bool
		truncatePending         bool
		compactionPending       bool
		closed                  bool
		closeDone               chan struct{}
		closeErr                error
	}

	fileSegment struct {
		Name           string
		Generation     uint64
		First          Position
		Last           Position
		ChecksumBefore uint64
		ChecksumAfter  uint64
		CommittedEnd   int64
		PhysicalBytes  int64
		Records        int
		Sealed         bool
		Standby        bool
		Replacement    bool
		Dirty          bool
		Transaction    bool
	}

	truncateCleanup struct {
		state            stateSlot
		obsolete         map[string]struct{}
		release          objectReleasePlan
		segmentsComplete bool
		objectsComplete  bool
	}

	fileIndexEntry struct {
		Position              Position
		Segment               uint64
		GroupOffset           int64
		TransactionGeneration uint64
		TransactionOffset     int64
		CommitGeneration      uint64
		FragmentStart         uint64
		FragmentCount         uint32
		GroupLength           uint32
		FrameOffset           uint32
		FrameLength           uint32
		ChecksumBefore        uint64
		ChecksumAfter         uint64
		ID                    string
		Digest                [sha256.Size]byte
		LogicalBytes          int64
		PayloadBytes          uint32
		Storage               recordStorage
	}

	fileFragment struct {
		Segment       uint64
		GroupOffset   int64
		GroupLength   uint32
		PayloadOffset uint32
		PayloadBytes  uint32
	}

	fileRecordIdentity struct {
		Position Position
		Digest   [sha256.Size]byte
	}
)

const (
	walInitialIndexRecords      = 256
	walReusableAppendGroupBytes = 2 << 20
	walIndexEntryBytes          = 160
	walFragmentIndexBytes       = 32
	walSegmentIndexBytes        = 128
)

func openFileLog(
	ctx context.Context,
	config FileConfig,
) (*fileLog, error) {
	normalized, err := normalizeFileWALConfig(config)
	if err != nil {
		return nil, err
	}

	progress := newRecoveryProgress(ctx, normalized)
	if err := progress.notify(Preparing, true); err != nil {
		return nil, err
	}

	metadata, err := openFileMetadata(ctx, normalized)
	if err != nil {
		return nil, err
	}

	objects, err := openFileObjects(metadata.fileSystem)
	if err != nil {
		return nil, errors.Join(err, metadata.close())
	}

	log := &fileLog{
		mutex:        newRankedContextMutex(lockrank.FileState),
		flushMutex:   newRankedContextMutex(lockrank.FileFlush),
		queueMutex:   lockrank.NewMutex(lockrank.FileQueue),
		stagingMutex: lockrank.NewMutex(lockrank.FileStaging),
		maintenanceMutex: lockrank.NewMutex(
			lockrank.FileMaintenance,
		),
		scanMutex: lockrank.NewMutex(lockrank.FileScan),
		metadata:  metadata,
		objects:   objects,
		config:    metadata.config,
		scanTokens: make(
			chan struct{},
			metadata.config.MaxConcurrentScans,
		),
		stagingTokens: make(
			chan struct{},
			metadata.config.MaxStagingRequests,
		),
		ids:       make(map[string]fileRecordIdentity),
		closeDone: make(chan struct{}),
		truncated: metadata.state.Truncated,
		last:      metadata.state.Last,
		checksum:  metadata.state.TruncatedChecksum,
	}
	log.queueCondition = sync.NewCond(&log.queueMutex)

	if err := progress.notify(Scanning, true); err != nil {
		return nil, errors.Join(err, objects.close(), metadata.close())
	}

	if err := log.recover(ctx, progress); err != nil {
		return nil, errors.Join(err, objects.close(), metadata.close())
	}

	if err := log.objects.recover(
		ctx,
		log.entries,
		log.config,
		progress,
	); err != nil {
		return nil, errors.Join(err, objects.close(), metadata.close())
	}

	log.recountSegmentStats()

	progress.report.Total = progress.report.Completed

	if err := progress.notify(Complete, true); err != nil {
		return nil, errors.Join(err, objects.close(), metadata.close())
	}

	return log, nil
}

func (l *fileLog) recover(
	ctx context.Context,
	progress *recoveryProgress,
) error {
	start, skipped, err := l.recoveryStart()
	if err != nil {
		return err
	}

	expectedPosition := Position(0)
	expectedChecksum := uint64(0)
	recoveredLast := Position(0)
	committedSegments := 0
	truncatedChecksumFound := l.truncated == 0
	sequenceStarted := false

	var transaction recoveryTransaction

	obsolete := append([]string(nil), skipped...)

	for descriptorIndex, descriptor := range l.metadata.segments {
		if err := ctx.Err(); err != nil {
			return err
		}

		if descriptorIndex < start {
			continue
		}

		segment := fileSegment{
			Name:           descriptor.Name,
			Generation:     descriptor.Generation,
			PhysicalBytes:  descriptor.Size,
			Standby:        descriptor.Standby,
			ChecksumBefore: descriptor.Header.ChecksumBefore,
			Transaction:    descriptor.Header.Transaction,
		}
		if descriptor.Standby {
			l.appendRecoveredSegment(segment)

			continue
		}

		if descriptor.Partial {
			if transaction.active {
				transaction.names = append(
					transaction.names,
					segment.Name,
				)
				if err := l.discardRecoveredTransaction(
					progress,
					transaction.startGeneration,
				); err != nil {
					return err
				}

				transaction = recoveryTransaction{}

				break
			}

			if descriptor.Generation != l.highestInitializedGeneration() {
				return l.corrupt("partial activation is not final")
			}

			file, err := l.metadata.fileSystem.open(
				segment.Name,
				os.O_RDWR,
			)
			if err != nil {
				return wrapError(
					"cannot open partial WAL activation",
					err,
				)
			}

			repairErr := l.resetEmptySegment(file, &segment)

			closeErr := l.metadata.fileSystem.closeFile(file)
			if err := errors.Join(repairErr, closeErr); err != nil {
				return err
			}

			l.appendRecoveredSegment(segment)

			continue
		}

		segment.First = descriptor.Header.First
		if !sequenceStarted {
			if segment.First > l.truncated+1 {
				return l.corrupt("first retained segment starts after the checkpoint")
			}

			if segment.First == 1 && segment.ChecksumBefore != 0 {
				return l.corrupt("initial segment checksum seed is non-zero")
			}

			expectedPosition = segment.First
			expectedChecksum = segment.ChecksumBefore
			sequenceStarted = true

			if segment.First == l.truncated+1 {
				if segment.ChecksumBefore !=
					l.metadata.state.TruncatedChecksum {
					return l.corrupt(
						"retained segment checksum seed does not match the checkpoint",
					)
				}

				truncatedChecksumFound = true
			}
		} else if transaction.active &&
			!descriptor.Header.Transaction {
			return l.corrupt(
				"normal segment interrupts an append transaction",
			)
		} else if !transaction.active &&
			(segment.First != expectedPosition ||
				segment.ChecksumBefore != expectedChecksum) {
			return l.corrupt("segment sequence or checksum seed mismatch")
		}

		finalActive := descriptorIndex ==
			l.lastInitializedDescriptor()
		if descriptor.Header.Transaction {
			file, err := l.metadata.fileSystem.open(
				segment.Name,
				os.O_RDWR,
			)
			if err != nil {
				return wrapError(
					"cannot open WAL transaction segment",
					err,
				)
			}

			committed, scanErr := l.scanTransactionSegment(
				ctx,
				progress,
				file,
				&segment,
				descriptor.Header,
				finalActive,
				true,
				&transaction,
				&expectedPosition,
				&expectedChecksum,
				&truncatedChecksumFound,
			)

			closeErr := l.metadata.fileSystem.closeFile(file)
			if err := errors.Join(scanErr, closeErr); err != nil {
				if errors.Is(err, errWALIncompleteTransaction) {
					if cleanupErr := l.discardRecoveredTransaction(
						progress,
						transaction.startGeneration,
					); cleanupErr != nil {
						return errors.Join(err, cleanupErr)
					}

					transaction = recoveryTransaction{}

					break
				}

				return err
			}

			l.appendRecoveredSegment(segment)

			if committed {
				committedSegments++

				recoveredLast = segment.Last
				if segment.Last <= l.truncated {
					obsolete = append(
						obsolete,
						transaction.names...,
					)
				}
			}

			if finalActive && transaction.active {
				if err := l.discardRecoveredTransaction(
					progress,
					transaction.startGeneration,
				); err != nil {
					return err
				}

				transaction = recoveryTransaction{}

				// Cleanup rewrites the descriptor slice and may publish a new
				// standby. Stop this recovery pass instead of allowing range
				// to observe the mutated backing array as another historical
				// descriptor.
				break
			}

			continue
		}

		if err := l.scanSegment(
			ctx,
			progress,
			&segment,
			finalActive,
			&expectedPosition,
			&expectedChecksum,
			&truncatedChecksumFound,
		); err != nil {
			return err
		}

		if segment.Standby {
			l.appendRecoveredSegment(segment)

			continue
		}

		if segment.Records == 0 {
			return l.corrupt("initialized segment has no committed group")
		}

		committedSegments++
		recoveredLast = segment.Last

		l.appendRecoveredSegment(segment)

		if segment.Last <= l.truncated {
			obsolete = append(obsolete, segment.Name)
		}
	}

	if transaction.active {
		return l.corrupt(
			"append transaction remains active after recovery",
		)
	}

	if committedSegments == 0 {
		if l.metadata.state.Last != l.metadata.state.Truncated ||
			l.metadata.state.HeadFirst != 0 ||
			l.metadata.state.HeadGeneration != 0 ||
			l.metadata.state.HeadOffset != 0 {
			return l.corrupt("non-empty state has no initialized segment")
		}

		l.last = l.truncated
		l.checksum = l.metadata.state.TruncatedChecksum

		return l.finishRecoveryCleanup(progress, obsolete)
	}

	if !truncatedChecksumFound {
		return l.corrupt("truncated checksum boundary is absent")
	}

	if l.metadata.state.Last > recoveredLast {
		return l.corrupt("state leads committed WAL data")
	}

	if recoveredLast < l.truncated {
		return l.corrupt("committed WAL ends before truncation")
	}

	l.last = recoveredLast
	l.checksum = expectedChecksum

	if l.metadata.state.HeadFirst != 0 {
		if l.metadata.state.HeadFirst != l.truncated+1 {
			return l.corrupt("state head position is inconsistent")
		}

		entry, found := l.entryAt(l.metadata.state.HeadFirst)
		headGeneration := entry.Segment

		headOffset := entry.GroupOffset
		if entry.TransactionGeneration != 0 {
			headGeneration = entry.TransactionGeneration
			headOffset = entry.TransactionOffset
		}

		if !found ||
			headGeneration != l.metadata.state.HeadGeneration ||
			uint64(headOffset) != l.metadata.state.HeadOffset {
			return l.corrupt("state head tuple does not select the live prefix")
		}
	}

	if len(l.entries) != int(l.last-l.truncated) {
		return l.corrupt("live index does not match the retained range")
	}

	if err := l.verifyCompactionRecoverySource(
		ctx,
		obsolete,
	); err != nil {
		return err
	}

	return l.finishRecoveryCleanup(progress, obsolete)
}

func (l *fileLog) recoveryStart() (int, []string, error) {
	firstInitialized := -1
	start := -1

	for index, descriptor := range l.metadata.segments {
		if descriptor.Standby {
			continue
		}

		if firstInitialized == -1 {
			firstInitialized = index
		}

		switch {
		case l.metadata.state.HeadGeneration != 0 &&
			descriptor.Generation == l.metadata.state.HeadGeneration:
			start = index

		case l.metadata.state.HeadGeneration == 0 &&
			l.metadata.state.Last != 0 &&
			!descriptor.Partial &&
			descriptor.Header.First <= l.metadata.state.Last:
			start = index
		}
	}

	if l.metadata.state.HeadGeneration != 0 && start == -1 {
		return 0, nil, l.corrupt("state head segment is absent")
	}

	if start == -1 {
		start = firstInitialized
	}

	if start == -1 {
		start = 0
	}

	skipped := make([]string, 0, start)
	for index := 0; index < start; index++ {
		descriptor := l.metadata.segments[index]
		if descriptor.Standby {
			return 0, nil, l.corrupt("standby precedes the recovery head")
		}

		skipped = append(skipped, descriptor.Name)
	}

	if len(skipped) != 0 && !l.metadata.state.CleanupPending {
		return 0, nil, l.corrupt("obsolete segments remain without cleanup state")
	}

	return start, skipped, nil
}

func (l *fileLog) finishRecoveryCleanup(
	progress *recoveryProgress,
	obsolete []string,
) error {
	if !l.metadata.state.CleanupPending {
		if len(obsolete) != 0 {
			return l.corrupt("obsolete segments remain after cleanup")
		}

		return nil
	}

	if err := progress.notify(Cleaning, true); err != nil {
		return err
	}

	removed := make(map[string]struct{}, len(obsolete))
	for _, name := range obsolete {
		if _, duplicate := removed[name]; duplicate {
			continue
		}

		if err := l.metadata.fileSystem.remove(name); err != nil {
			return wrapError(
				"cannot finish WAL checkpoint cleanup; retry recovery",
				err,
			)
		}

		removed[name] = struct{}{}
		if err := progress.removed(); err != nil {
			return err
		}
	}

	if len(removed) != 0 {
		if err := l.metadata.fileSystem.syncDirectory(); err != nil {
			return l.syncFailure(
				"cannot synchronize WAL checkpoint cleanup",
				err,
			)
		}
	}

	state := l.metadata.state
	if state.Generation == math.MaxUint64 {
		return ErrCapacity
	}

	state.Generation++

	state.CleanupPending = false

	state.FileGenerationHigh = l.metadata.fileGenerationHigh
	if err := l.metadata.publishState(state); err != nil {
		return err
	}

	l.metadata.state = state

	l.removeSegmentDescriptors(removed)

	return nil
}

func (l *fileLog) scanSegment(
	ctx context.Context,
	progress *recoveryProgress,
	segment *fileSegment,
	finalActive bool,
	expectedPosition *Position,
	expectedChecksum *uint64,
	truncatedChecksumFound *bool,
) (resultErr error) {
	file, err := l.metadata.fileSystem.open(segment.Name, os.O_RDWR)
	if err != nil {
		return wrapError("cannot open initialized WAL segment", err)
	}
	defer func() {
		if closeErr := l.metadata.fileSystem.closeFile(file); closeErr != nil {
			resultErr = errors.Join(
				resultErr,
				wrapError("cannot close recovered WAL segment", closeErr),
			)
		}
	}()

	offset := int64(walSegmentHeaderBytes)
	checksum := segment.ChecksumBefore
	records := 0

	var prefix [walGroupHeaderBytes]byte

	for offset < segment.PhysicalBytes {
		if err := ctx.Err(); err != nil {
			return err
		}

		remaining := segment.PhysicalBytes - offset
		if remaining < walGroupHeaderBytes {
			if !finalActive {
				return l.corrupt("short interior group header")
			}

			if err := l.repairTail(file, segment, offset); err != nil {
				return err
			}

			break
		}

		if err := l.metadata.fileSystem.readFullAt(file, prefix[:], offset); err != nil {
			return wrapError("cannot read WAL group header", err)
		}

		if allWALZero(prefix[:]) {
			zero, err := l.metadata.verifyAllZero(
				ctx,
				file,
				offset,
				segment.PhysicalBytes-offset,
			)
			if err != nil {
				return err
			}

			if !zero {
				return l.corrupt("non-zero data follows zero segment tail")
			}

			if !finalActive {
				return l.corrupt("zero-extended tail is not final")
			}

			if err := l.repairTail(file, segment, offset); err != nil {
				return err
			}

			break
		}

		if equalWALMagic(prefix[:4], walFooterMagic) {
			footerData := make([]byte, walSegmentFooterBytes)
			if segment.PhysicalBytes-offset < walSegmentFooterBytes {
				if !finalActive {
					return l.corrupt("short sealed footer")
				}

				if err := l.repairTail(file, segment, offset); err != nil {
					return err
				}

				break
			}

			if err := l.metadata.fileSystem.readFullAt(
				file,
				footerData,
				offset,
			); err != nil {
				return wrapError("cannot read sealed footer", err)
			}

			footer, err := decodeWALSegmentFooter(footerData)
			if err != nil ||
				footer.Last != segment.Last ||
				footer.Records != uint64(records) ||
				footer.Offset != uint64(offset) ||
				footer.FinalChecksum != checksum {
				return l.corrupt("sealed footer does not match segment")
			}

			if finalActive {
				if err := l.repairTail(file, segment, offset); err != nil {
					return err
				}

				break
			}

			segment.Sealed = true
			tailOffset := offset + walSegmentFooterBytes

			zero, err := l.metadata.verifyAllZero(
				ctx,
				file,
				tailOffset,
				segment.PhysicalBytes-tailOffset,
			)
			if err != nil {
				return err
			}

			if !zero {
				return l.corrupt("non-zero data follows sealed footer")
			}

			break
		}

		if !equalWALMagic(prefix[:4], walGroupMagic) {
			if !finalActive {
				return l.corrupt("invalid interior group magic")
			}

			if err := l.repairTail(file, segment, offset); err != nil {
				return err
			}

			break
		}

		groupBytes := int64(binary.BigEndian.Uint32(prefix[4:8]))
		if groupBytes < walGroupHeaderBytes+walRequestHeaderBytes+
			walFrameHeaderBytes+walDigestBytes+walChecksumBytes+
			walGroupTrailerBytes ||
			groupBytes > l.config.MaxCommitGroupBytes {
			return l.corrupt("group length exceeds configured bounds")
		}

		if groupBytes > segment.PhysicalBytes-offset {
			if !finalActive {
				return l.corrupt("incomplete group in sealed segment")
			}

			if err := l.repairTail(file, segment, offset); err != nil {
				return err
			}

			break
		}

		data := make([]byte, groupBytes)
		if err := l.metadata.fileSystem.readFullAt(file, data, offset); err != nil {
			return wrapError("cannot read complete WAL group", err)
		}

		group, err := decodeWALGroup(
			data,
			checksum,
			walFormatLimitsForConfig(l.config),
		)
		if err != nil {
			return l.corrupt("complete WAL group failed validation")
		}

		if group.First != *expectedPosition {
			return l.corrupt("group position sequence has a gap")
		}

		groupChecksum := checksum

		for _, decoded := range group.Records {
			groupChecksum = updateWALLogicalChecksumBytes(
				groupChecksum,
				decoded.Position,
				[]byte(decoded.ID),
				decoded.PayloadBytes,
				decoded.Digest[:],
			)

			if decoded.Position == l.truncated {
				if groupChecksum != l.metadata.state.TruncatedChecksum {
					return l.corrupt(
						"truncated checksum does not match the logical chain",
					)
				}

				*truncatedChecksumFound = true
			}

			if decoded.Position <= l.truncated {
				continue
			}

			if _, duplicate := l.ids[decoded.ID]; duplicate {
				return l.corrupt("duplicate live stable ID")
			}

			logicalBytes := int64(len(decoded.ID)) +
				int64(decoded.PayloadBytes)
			if logicalBytes > l.config.MaxBytes-l.logicalBytes ||
				len(l.entries) >= l.config.MaxRecords {
				return ErrCapacity
			}

			indexBytes := int64(walIndexEntryBytes + len(decoded.ID))
			if indexBytes > l.config.MaxIndexBytes-l.indexBytes {
				return ErrCapacity
			}

			entry := fileIndexEntry{
				Position:       decoded.Position,
				Segment:        segment.Generation,
				GroupOffset:    offset,
				GroupLength:    uint32(groupBytes),
				FrameOffset:    decoded.Offset,
				FrameLength:    decoded.Length,
				ChecksumBefore: group.ChecksumBefore,
				ChecksumAfter:  groupChecksum,
				ID:             decoded.ID,
				Digest:         decoded.Digest,
				LogicalBytes:   logicalBytes,
				PayloadBytes:   decoded.PayloadBytes,
				Storage:        decoded.Storage,
			}

			l.ensureEntryIndexCapacity(1)
			l.entries = append(l.entries, entry)
			l.ids[entry.ID] = fileRecordIdentity{
				Position: entry.Position,
				Digest:   entry.Digest,
			}
			l.logicalBytes += logicalBytes
			l.indexBytes += indexBytes

			if err := progress.recovered(logicalBytes); err != nil {
				return err
			}
		}

		checksum = group.ChecksumAfter
		*expectedChecksum = checksum
		*expectedPosition = group.Last + 1
		segment.Last = group.Last
		segment.ChecksumAfter = checksum
		records += len(group.Records)
		offset += groupBytes
	}

	segment.CommittedEnd = offset
	segment.Records = records

	if finalActive && records == 0 {
		return l.resetEmptySegment(file, segment)
	}

	if !finalActive && !segment.Sealed {
		return l.corrupt("interior segment is not sealed")
	}

	return nil
}

func (l *fileLog) resetEmptySegment(
	file *os.File,
	segment *fileSegment,
) error {
	if err := l.metadata.fileSystem.truncate(file, 0); err != nil {
		return wrapError("cannot reset empty active WAL segment", err)
	}

	if err := l.metadata.fileSystem.syncFile(file); err != nil {
		return l.syncFailure("cannot synchronize empty segment repair", err)
	}

	if l.config.Preallocation != PreallocateDisabled {
		if err := l.metadata.fileSystem.reserve(
			file,
			0,
			l.config.SegmentMaxBytes,
		); err != nil {
			if l.config.Preallocation == PreallocateRequired {
				return errors.Join(
					ErrCapability,
					wrapError(
						"cannot restore required standby allocation",
						err,
					),
				)
			}

			l.metadata.preallocationFailure++
		} else {
			l.metadata.preallocationSuccess++
		}
	}

	info, err := file.Stat()
	if err != nil {
		return wrapError("cannot inspect repaired standby segment", err)
	}

	*segment = fileSegment{
		Name:          segment.Name,
		Generation:    segment.Generation,
		PhysicalBytes: info.Size(),
		Standby:       true,
	}

	return nil
}

func (l *fileLog) repairTail(
	file *os.File,
	segment *fileSegment,
	committedEnd int64,
) error {
	if err := l.metadata.fileSystem.truncate(file, committedEnd); err != nil {
		return wrapError("cannot truncate torn WAL tail", err)
	}

	if err := l.metadata.fileSystem.syncFile(file); err != nil {
		return l.syncFailure("cannot synchronize WAL tail repair", err)
	}

	if l.config.Preallocation != PreallocateDisabled {
		if err := l.metadata.fileSystem.reserve(
			file,
			committedEnd,
			l.config.SegmentMaxBytes-committedEnd,
		); err != nil {
			if l.config.Preallocation == PreallocateRequired {
				return errors.Join(
					ErrCapability,
					wrapError(
						"cannot restore required segment allocation",
						err,
					),
				)
			}

			l.metadata.preallocationFailure++
		} else {
			l.metadata.preallocationSuccess++
		}
	}

	info, err := file.Stat()
	if err != nil {
		return wrapError("cannot inspect repaired WAL tail", err)
	}

	segment.PhysicalBytes = info.Size()
	segment.CommittedEnd = committedEnd
	segment.Sealed = false

	return nil
}

func (l *fileLog) commitPhysicalGroupLocked(
	requests []formatRequest,
	durability Durability,
) (decodedGroup, Durability, error) {
	logicalBytes := int64(0)
	indexBytes := int64(0)
	recordCount := 0

	for _, request := range requests {
		if len(request.Owned) != len(request.Records) {
			return decodedGroup{}, Written, internalError(
				"append identity ownership is incomplete",
			)
		}

		for index, record := range request.Records {
			owned := request.Owned[index]
			logicalBytes += int64(len(record.ID)) +
				int64(owned.PayloadBytes)
			indexBytes += int64(walIndexEntryBytes + len(record.ID))
			recordCount++
		}
	}

	// A gather write could avoid the contiguous encoded group, but partial
	// vector writes must preserve the exact commit-trailer boundary and
	// verified rollback semantics. Reuse one bounded buffer instead, clear it
	// before return, and keep the proven contiguous-write contract.
	encoded, decoded, err := encodeWALAppendGroup(
		l.appendBuffer,
		requests,
		l.last+1,
		l.checksum,
		walFormatLimitsForConfig(l.config),
	)
	if err != nil {
		return decodedGroup{}, Written, err
	}
	defer l.releaseAppendBuffer(encoded)

	data := encoded[walSegmentHeaderBytes:]

	if err := l.checkAppendDiskCapacity(
		len(data),
		recordCount,
	); err != nil {
		return decoded, Written, err
	}

	segment, activated, err := l.segmentForGroup(
		len(data),
		recordCount,
	)
	if err != nil {
		return decoded, Written, err
	}

	writeOffset := segment.CommittedEnd
	if activated {
		encodeWALSegmentHeaderInto(
			encoded[:walSegmentHeaderBytes],
			segmentHeader{
				WALID:          l.metadata.identity.ID,
				Generation:     segment.Generation,
				First:          decoded.First,
				ChecksumBefore: l.checksum,
			},
		)
		data = encoded
		writeOffset = 0
	}

	file, err := l.openActiveSegment(segment)
	if err != nil {
		return decoded, Written,
			wrapError("cannot open active WAL segment", err)
	}

	l.ensureEntryIndexCapacity(recordCount)

	oldEnd := segment.CommittedEnd
	if err := l.writeCommitData(
		file,
		data,
		writeOffset,
	); err != nil {
		rollbackErr := l.rollbackAppend(file, segment, oldEnd, activated)

		_ = l.closeActiveSegment()

		if rollbackErr != nil {
			return decoded, Written,
				errors.Join(
					ErrCommitUnknown,
					ErrSyncFailed,
					rollbackErr,
				)
		}

		return decoded, Written, wrapError(
			"cannot write WAL group; check storage health and free space",
			err,
		)
	}

	position := decoded.First

	for _, request := range requests {
		for _, owned := range request.Owned {
			l.ids[owned.ID] = fileRecordIdentity{
				Position: position,
				Digest:   owned.Digest,
			}
			position++
		}
	}

	wasStandby := segment.Standby
	segment.CommittedEnd = writeOffset + int64(len(data))
	l.setSegmentPhysicalBytes(segment, segment.CommittedEnd)
	segment.Last = decoded.Last
	segment.ChecksumAfter = decoded.ChecksumAfter
	segment.Records += recordCount
	segment.Standby = false

	if wasStandby {
		l.segmentCount++
	}

	l.activeBytes = segment.CommittedEnd
	l.activeRecords = segment.Records
	l.markSegmentDirty(segment)
	l.publishGroup(
		segment,
		requests,
		decoded,
		writeOffset+int64(len(data))-int64(decoded.Length),
		logicalBytes,
		indexBytes,
	)

	if durability == Synced {
		if err := l.syncDirtyThroughResponsive(
			segment.Generation,
			decoded.Last,
		); err != nil {
			return decoded, Written,
				errors.Join(ErrCommitUnknown, err)
		}
	}

	effective := Written
	if durability == Synced {
		effective = Synced
	}

	return decoded, effective, nil
}

func (l *fileLog) writeCommitData(
	file *os.File,
	data []byte,
	offset int64,
) error {
	// Commit leadership and append-prefix maintenance exclusion remain active,
	// but no uncommitted ID or logical position is published yet. Releasing
	// only the state mutex around the potentially large write keeps State,
	// Stats, and retained reads responsive without permitting another writer.
	l.mutex.Unlock()
	err := l.metadata.fileSystem.writeFullAt(file, data, offset)
	l.mutex.Lock()

	return err
}

func (l *fileLog) releaseAppendBuffer(buffer []byte) {
	// Clearing is an intentional memory-hygiene cost. Omitting it would save
	// one memory-bandwidth pass, but would keep copied record IDs and payloads
	// reachable from a long-lived WAL after Append returns.
	clear(buffer)

	if cap(buffer) >
		walSegmentHeaderBytes+walReusableAppendGroupBytes {
		return
	}

	l.appendBuffer = buffer[:0]
}

func (l *fileLog) segmentForGroup(
	groupBytes int,
	records int,
) (*fileSegment, bool, error) {
	active := l.activeSegment()
	required := int64(groupBytes + walSegmentFooterBytes)

	if int64(walSegmentHeaderBytes+groupBytes+walSegmentFooterBytes) >
		l.config.SegmentMaxBytes {
		return nil, false, ErrTooLarge
	}

	if active != nil &&
		!active.Transaction &&
		active.Records+records <= l.config.SegmentMaxRecords &&
		active.CommittedEnd+required <= l.config.SegmentMaxBytes {
		return active, false, nil
	}

	standby := l.standbySegment()
	if standby == nil {
		if walSegmentIndexBytes >
			l.config.MaxIndexBytes-l.indexBytes {
			return nil, false, ErrCapacity
		}

		if err := l.metadata.createStandby(); err != nil {
			return nil, false, err
		}

		descriptor := l.metadata.segments[len(l.metadata.segments)-1]
		l.appendRecoveredSegment(fileSegment{
			Name:          descriptor.Name,
			Generation:    descriptor.Generation,
			PhysicalBytes: descriptor.Size,
			Standby:       true,
		})
		active = l.activeSegment()
		standby = &l.segments[len(l.segments)-1]
	}

	if active != nil {
		if err := l.sealActive(active); err != nil {
			return nil, false, err
		}
	}

	standby.First = l.last + 1
	standby.ChecksumBefore = l.checksum
	standby.CommittedEnd = walSegmentHeaderBytes

	return standby, true, nil
}

func (l *fileLog) sealActive(segment *fileSegment) error {
	footer := encodeWALSegmentFooter(segmentFooter{
		Last:          segment.Last,
		Records:       uint64(segment.Records),
		Offset:        uint64(segment.CommittedEnd),
		FinalChecksum: segment.ChecksumAfter,
	})

	file, err := l.openActiveSegment(segment)
	if err != nil {
		return wrapError("cannot open active segment for rotation", err)
	}

	if err := l.metadata.fileSystem.writeFullAt(
		file,
		footer,
		segment.CommittedEnd,
	); err != nil {
		_ = l.closeActiveSegment()

		return wrapError("cannot seal active WAL segment", err)
	}

	segment.Sealed = true
	l.markSegmentDirty(segment)
	l.setSegmentPhysicalBytes(
		segment,
		segment.CommittedEnd+walSegmentFooterBytes,
	)
	l.activeBytes = 0
	l.activeRecords = 0

	if err := l.closeActiveSegment(); err != nil {
		return wrapError("cannot close sealed WAL segment", err)
	}

	return nil
}

func (l *fileLog) checkAppendDiskCapacity(
	groupBytes int,
	records int,
) error {
	active := l.activeSegment()
	required := int64(groupBytes + walSegmentFooterBytes)
	growth := int64(groupBytes)

	if active == nil ||
		active.Records+records > l.config.SegmentMaxRecords ||
		active.CommittedEnd+required > l.config.SegmentMaxBytes {
		growth += walSegmentHeaderBytes
		if active != nil {
			growth += walSegmentFooterBytes
		}
	}

	managedBytes := l.capacityManagedWALBytes()
	if managedBytes > l.config.MaxDiskBytes ||
		growth > l.config.MaxDiskBytes-managedBytes {
		return ErrCapacity
	}

	return nil
}

func (l *fileLog) appendRecoveredSegment(segment fileSegment) {
	l.segments = append(l.segments, segment)
	l.segmentBytes.Add(segment.PhysicalBytes)
	l.indexBytes += walSegmentIndexBytes
}

func (l *fileLog) setSegmentPhysicalBytes(
	segment *fileSegment,
	size int64,
) {
	l.segmentBytes.Add(size - segment.PhysicalBytes)
	segment.PhysicalBytes = size
}

func (l *fileLog) markSegmentDirty(segment *fileSegment) {
	if !segment.Dirty {
		l.dirtySegments++
		segment.Dirty = true
	}
}

func (l *fileLog) markSegmentClean(segment *fileSegment) {
	if segment.Dirty {
		l.dirtySegments--
		segment.Dirty = false
	}
}

func (l *fileLog) recountSegmentStats() {
	l.segmentBytes.Store(0)
	l.segmentCount = 0
	l.dirtySegments = 0
	l.activeBytes = 0
	l.activeRecords = 0

	for index := range l.segments {
		segment := &l.segments[index]

		l.segmentBytes.Add(segment.PhysicalBytes)

		if segment.Standby || segment.Replacement {
			continue
		}

		l.segmentCount++
		if segment.Dirty {
			l.dirtySegments++
		}

		if !segment.Sealed {
			l.activeBytes = segment.CommittedEnd
			l.activeRecords = segment.Records
		}
	}
}

func (l *fileLog) rollbackAppend(
	file *os.File,
	segment *fileSegment,
	oldEnd int64,
	activated bool,
) error {
	target := oldEnd
	if activated {
		target = 0
	}

	if err := l.metadata.fileSystem.truncate(file, target); err != nil {
		l.syncFailed = true

		return wrapError("cannot roll back failed WAL append", err)
	}

	if err := l.metadata.fileSystem.syncFile(file); err != nil {
		l.syncFailed = true

		return l.syncFailure("cannot synchronize WAL append rollback", err)
	}

	if err := l.validateRollback(file, segment, target, activated); err != nil {
		l.syncFailed = true

		return l.syncFailure(
			"cannot verify the restored WAL append boundary",
			err,
		)
	}

	if l.config.Preallocation != PreallocateDisabled {
		if err := l.metadata.fileSystem.reserve(
			file,
			target,
			l.config.SegmentMaxBytes-target,
		); err != nil {
			l.metadata.preallocationFailure++

			if l.config.Preallocation == PreallocateRequired {
				l.syncFailed = true

				return errors.Join(ErrCapability, err)
			}
		} else {
			l.metadata.preallocationSuccess++
		}
	}

	if activated {
		segment.First = 0
		segment.ChecksumBefore = 0
		segment.CommittedEnd = 0
		segment.Standby = true
	}

	return nil
}

func (l *fileLog) validateRollback(
	file *os.File,
	segment *fileSegment,
	target int64,
	activated bool,
) error {
	info, err := file.Stat()
	if err != nil {
		return wrapError("cannot inspect restored WAL append boundary", err)
	}

	if info.Size() != target {
		return internalError("restored append boundary has the wrong size")
	}

	if activated {
		if target != 0 {
			return internalError("reset standby has a non-zero boundary")
		}

		return nil
	}

	if target <= walSegmentHeaderBytes || l.last == l.truncated {
		return internalError("prior committed group is absent")
	}

	entry, found := l.entryAt(l.last)
	if !found ||
		entry.Segment != segment.Generation ||
		entry.GroupOffset+int64(entry.GroupLength) != target {
		return internalError(
			"prior committed group boundary is inconsistent",
		)
	}

	data := make([]byte, entry.GroupLength)
	if err := l.metadata.fileSystem.readFullAt(
		file,
		data,
		entry.GroupOffset,
	); err != nil {
		return wrapError("cannot read prior committed WAL group", err)
	}

	group, err := decodeWALGroup(
		data,
		entry.ChecksumBefore,
		walFormatLimitsForConfig(l.config),
	)
	if err != nil ||
		group.Last != segment.Last ||
		group.ChecksumAfter != segment.ChecksumAfter {
		return internalError("prior committed group failed validation")
	}

	return nil
}

func (l *fileLog) syncDirtyThrough(
	generation uint64,
	through Position,
) error {
	return l.syncDirtyThroughMode(
		context.Background(),
		generation,
		through,
		false,
	)
}

func (l *fileLog) syncDirtyThroughResponsive(
	generation uint64,
	through Position,
) error {
	return l.syncDirtyThroughMode(
		context.Background(),
		generation,
		through,
		true,
	)
}

func (l *fileLog) syncDirtyThroughContext(
	ctx context.Context,
	generation uint64,
	through Position,
) error {
	return l.syncDirtyThroughMode(ctx, generation, through, false)
}

func (l *fileLog) syncDirtyThroughMode(
	ctx context.Context,
	generation uint64,
	through Position,
	responsive bool,
) error {
	if generation == 0 ||
		l.segmentForPosition(through) != generation {
		return l.corrupt("synchronization target has no segment")
	}

	for index := range l.segments {
		if err := ctx.Err(); err != nil {
			return err
		}

		segment := &l.segments[index]
		if segment.Standby || !segment.Dirty ||
			segment.Generation > generation {
			continue
		}

		file, borrowed, err := l.openSegmentForSync(segment)
		if err != nil {
			l.syncFailed = true

			return l.syncFailure("cannot open dirty WAL segment", err)
		}

		var syncErr, closeErr error

		if responsive {
			// The caller must retain commit leadership or an explicit flush
			// barrier plus maintenance exclusion. No writer may mutate segment
			// descriptors while the state mutex is released.
			l.mutex.Unlock()

			syncErr = l.metadata.fileSystem.syncFile(file)
			if borrowed {
				closeErr = l.metadata.fileSystem.closeFile(file)
			}

			l.mutex.Lock()
		} else {
			syncErr = l.metadata.fileSystem.syncFile(file)
			if borrowed {
				closeErr = l.metadata.fileSystem.closeFile(file)
			}
		}

		if err := errors.Join(syncErr, closeErr); err != nil {
			l.syncFailed = true

			return l.syncFailure("cannot synchronize dirty WAL segment", err)
		}

		l.markSegmentClean(segment)
	}

	if err := ctx.Err(); err != nil {
		return err
	}

	l.syncedThrough = max(l.syncedThrough, through)

	return nil
}

func (l *fileLog) publishGroup(
	segment *fileSegment,
	requests []formatRequest,
	group decodedGroup,
	groupOffset int64,
	logicalBytes, indexBytes int64,
) {
	position := group.First
	frameOffset := walGroupHeaderBytes

	for _, request := range requests {
		frameOffset += walRequestHeaderBytes

		for recordIndex, record := range request.Records {
			owned := request.Owned[recordIndex]
			frameLength := walFrameEncodedSize(
				record,
				owned.Storage,
				owned.PayloadBytes,
			)

			entry := fileIndexEntry{
				Position:       position,
				Segment:        segment.Generation,
				GroupOffset:    groupOffset,
				GroupLength:    group.Length,
				FrameOffset:    uint32(frameOffset),
				FrameLength:    uint32(frameLength),
				ChecksumBefore: group.ChecksumBefore,
				ChecksumAfter:  owned.ChecksumAfter,
				ID:             owned.ID,
				Digest:         owned.Digest,
				LogicalBytes: int64(len(record.ID)) +
					int64(owned.PayloadBytes),
				Storage:      owned.Storage,
				PayloadBytes: owned.PayloadBytes,
			}
			l.entries = append(l.entries, entry)
			l.ids[entry.ID] = fileRecordIdentity{
				Position: entry.Position,
				Digest:   entry.Digest,
			}

			position++
			frameOffset += frameLength
		}

		l.objects.retain(request.Owned)
	}

	l.logicalBytes += logicalBytes
	l.indexBytes += indexBytes
	l.last = group.Last
	l.checksum = group.ChecksumAfter
}

func (l *fileLog) ensureEntryIndexCapacity(additional int) {
	required := len(l.entries) + additional
	if required <= cap(l.entries) {
		return
	}

	// Reserving MaxRecords at open would avoid geometric copies but retain the
	// configured worst case even for small WALs. Grow from a bounded floor and
	// cap each step at both record and index-byte limits instead.
	limit := l.config.MaxRecords

	byteLimit := l.config.MaxIndexBytes / walIndexEntryBytes
	if byteLimit < int64(limit) {
		limit = int(byteLimit)
	}

	capacity := walInitialIndexRecords

	if current := cap(l.entries); current > 0 {
		if current > limit/2 {
			capacity = limit
		} else {
			capacity = current * 2
		}
	}

	capacity = min(max(capacity, required), limit)
	entries := make([]fileIndexEntry, len(l.entries), capacity)
	copy(entries, l.entries)
	l.entries = entries
}

func (l *fileLog) ensureFragmentIndexCapacity(additional int) {
	required := len(l.fragments) + additional
	if required <= cap(l.fragments) {
		return
	}

	limit := int(
		l.config.MaxIndexBytes / walFragmentIndexBytes,
	)
	capacity := walInitialIndexRecords

	if current := cap(l.fragments); current > 0 {
		if current > limit/2 {
			capacity = limit
		} else {
			capacity = current * 2
		}
	}

	capacity = min(max(capacity, required), limit)
	fragments := make([]fileFragment, len(l.fragments), capacity)
	copy(fragments, l.fragments)
	l.fragments = fragments
}

func (l *fileLog) resolveRetry(
	records []Record,
	owned []ownedRecord,
	durability Durability,
) (Position, bool, error) {
	if len(owned) != len(records) {
		return 0, false, internalError("retry ownership is incomplete")
	}

	found := 0

	var first Position

	for index, record := range records {
		identity, exists := l.ids[record.ID]
		if !exists {
			continue
		}

		found++

		if index == 0 {
			first = identity.Position
		} else if first == 0 ||
			identity.Position != first+Position(index) {
			return 0, false, ErrConflict
		}

		if identity.Digest != owned[index].Digest {
			return 0, false, ErrConflict
		}

		stored, err := l.readRecord(identity.Position)
		if err != nil {
			return 0, false, err
		}

		if stored.ID != record.ID ||
			!bytes.Equal(stored.Data, record.Data) {
			return 0, false, ErrConflict
		}
	}

	if found == 0 {
		return 0, false, nil
	}

	if found != len(records) || first == 0 {
		return 0, false, ErrConflict
	}

	if durability == Synced && first+Position(len(records))-1 >
		l.syncedThrough {
		through := first + Position(len(records)) - 1

		generation := l.segmentForPosition(through)
		if generation == 0 {
			return 0, false, l.corrupt("retry barrier has no segment")
		}

		if err := l.syncDirtyThroughResponsive(
			generation,
			through,
		); err != nil {
			return 0, false, err
		}
	}

	return first, true, nil
}

func (l *fileLog) readRecord(position Position) (Record, error) {
	entry, found := l.entryAt(position)
	if !found {
		return Record{}, ErrInvalidPosition
	}

	if entry.FragmentCount != 0 {
		fragmented, err := l.fragmentedRecordLocked(entry)
		if err != nil {
			return Record{}, err
		}

		record, _, err := l.readFragmentedRecord(
			context.Background(),
			fragmented,
			nil,
		)
		if err != nil {
			return Record{}, err
		}

		return record, nil
	}

	segment := l.segmentByGeneration(entry.Segment)
	if segment == nil {
		return Record{}, l.corrupt("indexed segment is absent")
	}

	file, err := l.metadata.fileSystem.open(segment.Name, os.O_RDONLY)
	if err != nil {
		return Record{}, wrapError("cannot open indexed WAL segment", err)
	}

	data := make([]byte, entry.GroupLength)
	readErr := l.metadata.fileSystem.readFullAt(
		file,
		data,
		entry.GroupOffset,
	)

	closeErr := l.metadata.fileSystem.closeFile(file)
	if err := errors.Join(readErr, closeErr); err != nil {
		return Record{}, wrapError("cannot read indexed WAL group", err)
	}

	group, err := decodeWALGroup(
		data,
		entry.ChecksumBefore,
		walFormatLimitsForConfig(l.config),
	)
	if err != nil {
		return Record{}, l.corrupt("indexed WAL group failed validation")
	}

	for _, record := range group.Records {
		if record.Position == position {
			if record.Storage != entry.Storage ||
				record.PayloadBytes != entry.PayloadBytes ||
				record.Digest != entry.Digest {
				return Record{},
					l.corrupt("indexed WAL record differs from its frame")
			}

			if record.Storage == recordBlob {
				payload := make([]byte, record.PayloadBytes)
				if err := l.readObjectData(
					context.Background(),
					objectKey{
						Digest: record.Digest,
						Bytes:  record.PayloadBytes,
					},
					payload,
				); err != nil {
					return Record{}, err
				}

				return Record{
					ID:   record.ID,
					Data: payload,
				}, nil
			}

			return Record{
				ID:   record.ID,
				Data: append([]byte(nil), record.Data...),
			}, nil
		}
	}

	return Record{}, l.corrupt("indexed WAL position is absent from its group")
}

func (l *fileLog) read(
	ctx context.Context,
	after Position,
	limit int,
) ([]Entry, error) {
	if limit <= 0 {
		return []Entry{}, nil
	}

	entries := make([]Entry, 0, limit)

	err := l.scan(ctx, after, limit, func(entry Entry) error {
		entry.Record.Data = append([]byte(nil), entry.Record.Data...)
		entries = append(entries, entry)

		return nil
	})
	if err != nil {
		return nil, err
	}

	return entries, nil
}

func (l *fileLog) flush(
	ctx context.Context,
	durability Durability,
) (Position, error) {
	if err := ctx.Err(); err != nil {
		return 0, err
	}

	if !l.supports(durability) {
		return 0, ErrUnsupportedDurability
	}

	responsive := durability == Synced
	if responsive {
		if err := l.flushMutex.LockContext(ctx); err != nil {
			return 0, err
		}
		defer l.flushMutex.Unlock()

		if !l.maintenanceMutex.TryLock() {
			return 0, ErrBackpressure
		}
		defer l.maintenanceMutex.Unlock()

		if err := l.beginFlushBarrier(ctx); err != nil {
			return 0, err
		}
		defer l.endFlushBarrier()
	} else {
		if err := l.waitForAppendPrefix(ctx); err != nil {
			return 0, err
		}
	}

	if err := l.mutex.LockContext(ctx); err != nil {
		return 0, err
	}
	defer l.mutex.Unlock()

	if l.closed {
		return 0, ErrClosed
	}

	if l.syncFailed {
		return 0, ErrSyncFailed
	}

	if l.stateUncertain || l.truncatePending || l.compactionPending {
		return 0, ErrUncertain
	}

	watermark := l.last
	if durability == Written || watermark == l.truncated {
		return watermark, nil
	}

	generation := l.segmentForPosition(watermark)
	if generation == 0 {
		return watermark, l.corrupt("flush watermark has no segment")
	}

	// Admission already succeeded, so the storage barrier must ignore later
	// cancellation. Do not derive context.WithoutCancel unless a callback
	// actually consumes it; the wrapper would otherwise add a hot-path
	// allocation without changing behavior.
	if err := l.syncDirtyThroughResponsive(
		generation,
		watermark,
	); err != nil {
		return watermark, err
	}

	return watermark, nil
}

func (l *fileLog) truncate(
	ctx context.Context,
	through Position,
	durability Durability,
) error {
	if err := ctx.Err(); err != nil {
		return err
	}

	if !l.supports(durability) {
		return ErrUnsupportedDurability
	}

	if !l.maintenanceMutex.TryLock() {
		return ErrBackpressure
	}
	defer l.maintenanceMutex.Unlock()

	if err := l.waitForAppendPrefix(ctx); err != nil {
		return err
	}

	if err := l.mutex.LockContext(ctx); err != nil {
		return err
	}
	defer l.mutex.Unlock()

	if l.closed {
		return ErrClosed
	}

	if l.syncFailed {
		return ErrSyncFailed
	}

	if l.stateUncertain {
		return ErrUncertain
	}

	if l.compactionPending {
		return ErrUncertain
	}

	if l.truncatePending {
		if through != l.metadata.state.Truncated {
			return ErrUncertain
		}

		if err := ctx.Err(); err != nil {
			return err
		}

		return l.resumeTruncateCleanup()
	}

	if through <= l.truncated {
		if durability == Synced && l.last > l.truncated {
			generation := l.segmentForPosition(l.last)
			if generation == 0 {
				return l.corrupt("truncate barrier has no retained segment")
			}

			if err := ctx.Err(); err != nil {
				return err
			}

			return l.syncDirtyThrough(generation, l.last)
		}

		return nil
	}

	if through > l.last {
		return ErrInvalidPosition
	}

	if durability == Synced && through < l.last {
		generation := l.segmentForPosition(l.last)
		if generation == 0 {
			return l.corrupt("retained truncate watermark has no segment")
		}

		if err := l.syncDirtyThroughContext(
			ctx,
			generation,
			l.last,
		); err != nil {
			return err
		}
	}

	if err := ctx.Err(); err != nil {
		return err
	}

	boundary, found := l.entryAt(through)
	if !found {
		return l.corrupt("truncate checksum boundary is absent")
	}

	obsolete := make(map[string]struct{})

	for _, segment := range l.segments {
		if err := ctx.Err(); err != nil {
			return err
		}

		if !segment.Standby &&
			(segment.Records != 0 || segment.Transaction) &&
			segment.Last <= through {
			obsolete[segment.Name] = struct{}{}
		}
	}

	releasePlan, err := l.objects.planRelease(l.entries, through)
	if err != nil {
		return err
	}

	state, err := l.nextTruncateState(
		through,
		boundary.ChecksumAfter,
		len(obsolete) != 0 || len(releasePlan.Remove) != 0,
	)
	if err != nil {
		return err
	}

	// This is the checkpoint-publication admission point. Cancellation before
	// it leaves the logical prefix unchanged. Once publication starts, the
	// caller receives its definite storage outcome even if ctx is canceled.
	// Future context-aware callbacks below this point must receive
	// context.WithoutCancel(ctx) at invocation time so values survive without
	// retaining cancellation.
	if err := ctx.Err(); err != nil {
		return err
	}

	if err := l.metadata.publishState(state); err != nil {
		l.stateUncertain = true
		if errors.Is(err, ErrSyncFailed) {
			l.syncFailed = true
		}

		return errors.Join(ErrTruncateUnknown, err)
	}

	l.metadata.state = state
	l.truncatePending = true

	if !state.CleanupPending {
		if err := l.objects.completeRelease(releasePlan); err != nil {
			return errors.Join(ErrTruncateUnknown, err)
		}

		l.applyTruncatedPrefix(through)
		l.truncatePending = false

		return nil
	}

	l.applyTruncatedPrefix(through)
	l.truncateCleanup = &truncateCleanup{
		state:    state,
		obsolete: obsolete,
		release:  releasePlan,
	}

	return l.completeTruncateCleanup(l.truncateCleanup)
}

func (l *fileLog) resumeTruncateCleanup() error {
	state := l.metadata.state
	if !state.CleanupPending {
		if state.Truncated > l.truncated {
			l.applyTruncatedPrefix(state.Truncated)
		}

		l.truncatePending = false

		return nil
	}

	if l.truncateCleanup != nil {
		return l.completeTruncateCleanup(l.truncateCleanup)
	}

	obsolete := make(map[string]struct{})

	for _, segment := range l.segments {
		if !segment.Standby &&
			(segment.Records != 0 || segment.Transaction) &&
			segment.Last <= state.Truncated {
			obsolete[segment.Name] = struct{}{}
		}
	}

	releasePlan, err := l.objects.planRelease(
		l.entries,
		state.Truncated,
	)
	if err != nil {
		return err
	}

	if state.Truncated > l.truncated {
		l.applyTruncatedPrefix(state.Truncated)
	}

	l.truncateCleanup = &truncateCleanup{
		state:    state,
		obsolete: obsolete,
		release:  releasePlan,
	}

	return l.completeTruncateCleanup(l.truncateCleanup)
}

func (l *fileLog) completeTruncateCleanup(
	cleanup *truncateCleanup,
) error {
	// The durable cleanup-pending checkpoint and in-memory logical prefix are
	// already published. Append and maintenance admission therefore fail
	// closed, while State, Stats, and reads beyond the new prefix remain
	// responsive during potentially slow unlink and directory barriers.
	l.mutex.Unlock()
	err := l.runTruncateCleanup(cleanup)
	l.mutex.Lock()

	if err != nil {
		if errors.Is(err, ErrSyncFailed) {
			l.syncFailed = true
		}

		return err
	}

	l.removeSegmentDescriptors(cleanup.obsolete)

	state := cleanup.state
	if state.Generation == math.MaxUint64 {
		return ErrCapacity
	}

	state.Generation++
	state.CleanupPending = false

	if err := l.metadata.publishState(state); err != nil {
		if errors.Is(err, ErrSyncFailed) {
			l.syncFailed = true
		}

		return errors.Join(ErrTruncateUnknown, err)
	}

	l.metadata.state = state
	l.truncatePending = false
	l.truncateCleanup = nil

	return nil
}

func (l *fileLog) runTruncateCleanup(
	cleanup *truncateCleanup,
) error {
	if cleanup == nil {
		return internalError("truncate cleanup plan is absent")
	}

	if !cleanup.segmentsComplete {
		if err := l.cleanupTruncatedSegments(cleanup.obsolete); err != nil {
			return err
		}

		cleanup.segmentsComplete = true
	}

	if !cleanup.objectsComplete {
		if err := l.objects.completeRelease(cleanup.release); err != nil {
			return errors.Join(
				ErrTruncateUnknown,
				wrapError(
					"truncate checkpoint is committed but payload cleanup is incomplete; retry the same truncation",
					err,
				),
			)
		}

		cleanup.objectsComplete = true
	}

	return nil
}

func (l *fileLog) cleanupTruncatedSegments(
	obsolete map[string]struct{},
) error {
	if err := l.closeActiveIfRemoved(obsolete); err != nil {
		return errors.Join(
			ErrTruncateUnknown,
			wrapError(
				"truncate checkpoint is committed but active segment cleanup is incomplete; retry the same truncation",
				err,
			),
		)
	}

	for name := range obsolete {
		if err := l.metadata.fileSystem.remove(name); err != nil &&
			!errors.Is(err, os.ErrNotExist) {
			return errors.Join(
				ErrTruncateUnknown,
				wrapError(
					"truncate checkpoint is committed but segment cleanup is incomplete; retry the same truncation",
					err,
				),
			)
		}
	}

	if len(obsolete) != 0 {
		if err := l.metadata.fileSystem.syncDirectory(); err != nil {
			return errors.Join(
				ErrTruncateUnknown,
				l.syncFailure(
					"cannot synchronize truncated segment cleanup",
					err,
				),
			)
		}
	}

	return nil
}

func (l *fileLog) nextTruncateState(
	through Position,
	checksum uint64,
	cleanupPending bool,
) (stateSlot, error) {
	state := l.metadata.state
	if state.Generation == math.MaxUint64 {
		return stateSlot{}, ErrCapacity
	}

	state.Generation++
	state.Last = l.last
	state.Truncated = through
	state.TruncatedChecksum = checksum
	state.FileGenerationHigh = l.metadata.fileGenerationHigh
	state.CleanupPending = cleanupPending

	if through == l.last {
		state.HeadFirst = 0
		state.HeadGeneration = 0
		state.HeadOffset = 0

		return state, nil
	}

	head, found := l.entryAt(through + 1)
	if !found {
		return stateSlot{}, l.corrupt("truncate head position is absent")
	}

	state.HeadFirst = through + 1
	if head.TransactionGeneration != 0 {
		state.HeadGeneration = head.TransactionGeneration
		state.HeadOffset = uint64(head.TransactionOffset)
	} else {
		state.HeadGeneration = head.Segment
		state.HeadOffset = uint64(head.GroupOffset)
	}

	return state, nil
}

func (l *fileLog) applyTruncatedPrefix(through Position) {
	remove := 0

	fragmentThrough := l.fragmentBase
	for remove < len(l.entries) &&
		l.entries[remove].Position <= through {
		entry := l.entries[remove]
		delete(l.ids, entry.ID)
		l.logicalBytes -= entry.LogicalBytes

		l.indexBytes -= int64(walIndexEntryBytes + len(entry.ID))
		if entry.FragmentCount != 0 {
			l.indexBytes -= int64(
				entry.FragmentCount * walFragmentIndexBytes,
			)
			fragmentThrough = max(
				fragmentThrough,
				entry.FragmentStart+uint64(entry.FragmentCount),
			)
		}

		remove++
	}

	copy(l.entries, l.entries[remove:])
	clear(l.entries[len(l.entries)-remove:])

	l.entries = l.entries[:len(l.entries)-remove]
	if fragmentThrough > l.fragmentBase {
		fragments := int(fragmentThrough - l.fragmentBase)
		copy(l.fragments, l.fragments[fragments:])
		clear(l.fragments[len(l.fragments)-fragments:])
		l.fragments = l.fragments[:len(l.fragments)-fragments]
		l.fragmentBase = fragmentThrough
	}

	l.truncated = through
}

func (l *fileLog) removeSegmentDescriptors(
	removed map[string]struct{},
) {
	if len(removed) == 0 {
		return
	}

	segments := l.segments[:0]
	for _, segment := range l.segments {
		if _, found := removed[segment.Name]; found {
			l.segmentBytes.Add(-segment.PhysicalBytes)
			l.indexBytes -= walSegmentIndexBytes

			if !segment.Standby && !segment.Replacement {
				l.segmentCount--
			}

			if segment.Dirty {
				l.dirtySegments--
			}

			if !segment.Standby &&
				!segment.Replacement &&
				!segment.Sealed {
				l.activeBytes = 0
				l.activeRecords = 0
			}

			continue
		}

		segments = append(segments, segment)
	}

	clear(l.segments[len(segments):])
	l.segments = segments

	descriptors := l.metadata.segments[:0]
	for _, descriptor := range l.metadata.segments {
		if _, found := removed[descriptor.Name]; found {
			continue
		}

		descriptors = append(descriptors, descriptor)
	}

	clear(l.metadata.segments[len(descriptors):])
	l.metadata.segments = descriptors
}

func (l *fileLog) state() State {
	l.mutex.Lock()
	defer l.mutex.Unlock()

	return State{
		Truncated: l.truncated,
		Last:      l.last,
	}
}

func (l *fileLog) supports(durability Durability) bool {
	return durability == Written ||
		durability == Synced && supportsWALSynced()
}

func (l *fileLog) activeSegment() *fileSegment {
	for index := len(l.segments) - 1; index >= 0; index-- {
		segment := &l.segments[index]
		if !segment.Standby &&
			!segment.Replacement &&
			!segment.Sealed {
			return segment
		}
	}

	return nil
}

func (l *fileLog) standbySegment() *fileSegment {
	if len(l.segments) == 0 {
		return nil
	}

	segment := &l.segments[len(l.segments)-1]
	if segment.Standby && !segment.Replacement {
		return segment
	}

	return nil
}

func (l *fileLog) segmentByGeneration(
	generation uint64,
) *fileSegment {
	for index := range l.segments {
		if l.segments[index].Generation == generation {
			return &l.segments[index]
		}
	}

	return nil
}

func (l *fileLog) segmentForPosition(position Position) uint64 {
	entry, found := l.entryAt(position)
	if !found {
		return 0
	}

	if entry.CommitGeneration != 0 {
		return entry.CommitGeneration
	}

	return entry.Segment
}

func (l *fileLog) entryAt(
	position Position,
) (fileIndexEntry, bool) {
	if position <= l.truncated || position > l.last {
		return fileIndexEntry{}, false
	}

	index := int(position - l.truncated - 1)
	if index < 0 || index >= len(l.entries) {
		return fileIndexEntry{}, false
	}

	entry := l.entries[index]

	return entry, entry.Position == position
}

func (l *fileLog) highestInitializedGeneration() uint64 {
	var generation uint64

	for _, segment := range l.metadata.segments {
		if !segment.Standby {
			generation = max(generation, segment.Generation)
		}
	}

	return generation
}

func (l *fileLog) lastInitializedDescriptor() int {
	for index := len(l.metadata.segments) - 1; index >= 0; index-- {
		segment := l.metadata.segments[index]
		if !segment.Standby && !segment.Partial {
			return index
		}
	}

	return -1
}

func (l *fileLog) corrupt(reason string) error {
	return errors.Join(
		ErrCorrupt,
		errors.New("wal: "+reason),
	)
}

func (l *fileLog) syncFailure(
	message string,
	cause error,
) error {
	l.syncFailed = true

	return errors.Join(
		ErrSyncFailed,
		wrapError(message+"; preserve the WAL epoch", cause),
	)
}

func (l *fileLog) close() error {
	l.queueMutex.Lock()
	if l.admissionClosed {
		done := l.closeDone
		l.queueMutex.Unlock()
		<-done

		return l.closeErr
	}
	l.queueMutex.Unlock()

	if !l.maintenanceMutex.TryLock() {
		return ErrBackpressure
	}
	defer l.maintenanceMutex.Unlock()

	l.stagingMutex.Lock()
	if l.preparingRequests != 0 {
		l.stagingMutex.Unlock()

		return ErrBackpressure
	}

	l.stagingClosed = true
	l.stagingMutex.Unlock()

	l.scanMutex.Lock()
	if l.activeScans != 0 {
		l.scanMutex.Unlock()
		l.stagingMutex.Lock()
		l.stagingClosed = false
		l.stagingMutex.Unlock()

		return ErrBackpressure
	}

	l.scanAdmissionClosed = true
	l.scanMutex.Unlock()

	l.queueMutex.Lock()
	if l.admissionClosed {
		done := l.closeDone
		l.queueMutex.Unlock()
		<-done

		return l.closeErr
	}

	if l.commitLeader {
		l.queueMutex.Unlock()

		l.scanMutex.Lock()
		l.scanAdmissionClosed = false
		l.scanMutex.Unlock()

		l.stagingMutex.Lock()
		l.stagingClosed = false
		l.stagingMutex.Unlock()

		return ErrBackpressure
	}

	l.admissionClosed = true

	for _, request := range l.pending {
		request.err = ErrClosed
		request.done = true

		l.pendingRequests--
		l.pendingBytes -= request.borrowedBytes
	}

	clear(l.pending)
	l.pending = nil
	l.queueCondition.Broadcast()

	l.queueMutex.Unlock()

	l.mutex.Lock()
	l.closed = true

	activeErr := l.closeActiveSegment()
	if activeErr != nil {
		activeErr = wrapError(
			"cannot close active WAL segment",
			activeErr,
		)
	}

	clear(l.appendBuffer)
	l.appendBuffer = nil
	l.mutex.Unlock()

	err := errors.Join(activeErr, l.objects.close(), l.metadata.close())

	l.queueMutex.Lock()
	l.closeErr = err
	close(l.closeDone)
	l.queueMutex.Unlock()

	return err
}

func (l *fileLog) openActiveSegment(
	segment *fileSegment,
) (*os.File, error) {
	if l.activeFile != nil {
		if l.activeFileGeneration != segment.Generation {
			return nil, l.corrupt(
				"active descriptor generation does not match the append segment",
			)
		}

		return l.activeFile, nil
	}

	file, err := l.metadata.fileSystem.open(segment.Name, os.O_RDWR)
	if err != nil {
		return nil, err
	}

	l.activeFile = file
	l.activeFileGeneration = segment.Generation

	return file, nil
}

func (l *fileLog) openSegmentForSync(
	segment *fileSegment,
) (*os.File, bool, error) {
	if l.activeFile != nil &&
		l.activeFileGeneration == segment.Generation {
		return l.activeFile, false, nil
	}

	file, err := l.metadata.fileSystem.open(segment.Name, os.O_RDWR)
	if err != nil {
		return nil, false, err
	}

	return file, true, nil
}

func (l *fileLog) closeActiveSegment() error {
	if l.activeFile == nil {
		return nil
	}

	file := l.activeFile
	l.activeFile = nil
	l.activeFileGeneration = 0

	return l.metadata.fileSystem.closeFile(file)
}

func (l *fileLog) closeActiveIfRemoved(
	removed map[string]struct{},
) error {
	if l.activeFile == nil {
		return nil
	}

	segment := l.segmentByGeneration(l.activeFileGeneration)
	if segment == nil {
		return l.corrupt("active descriptor segment is absent")
	}

	if _, found := removed[segment.Name]; !found {
		return nil
	}

	return l.closeActiveSegment()
}

func invokeWALScanCallback(
	visit func(Entry) error,
	entry Entry,
) (err error) {
	defer func() {
		if recovered := recover(); recovered != nil {
			err = newCallbackPanicError("WAL scan callback", recovered)
		}
	}()

	return visit(entry)
}