file.go

package wal

import (
	"context"
	"time"
)

type (
	// FileConfig defines one segmented filesystem WAL and its hard bounds.
	// Zero-valued limits select the bounded defaults documented by this
	// package. Recovery, capacity, physical-layout, and preallocation fields
	// form a profile persisted in the WAL. Reopen may keep or strengthen that
	// profile; weakening it returns ErrConfigDowngrade before recovery mutates
	// storage. Read, queue, staging, delay, callback, and progress fields
	// remain runtime tuning and may change on every open.
	FileConfig struct {
		// Directory is the required exclusively owned WAL directory.
		Directory string
		// MaxBytes bounds retained ID and payload bytes. Zero defaults to
		// 1 GiB.
		MaxBytes int64
		// MaxDiskBytes bounds managed file lengths plus active staging and
		// compaction reservations. Zero derives a bounded value from MaxBytes,
		// MaxStagingBytes, four segments, the metadata reserve, and fixed
		// metadata.
		MaxDiskBytes int64
		// MaxRecords bounds retained records. Zero defaults to 100,000.
		MaxRecords int
		// MaxRecordBytes bounds one logical payload across all fragments.
		// Zero defaults to 1 MiB.
		MaxRecordBytes int
		// MaxIDBytes bounds one stable ID. Zero defaults to 1 KiB.
		MaxIDBytes int
		// MaxReadEntries bounds one Read or Scan call. Zero defaults to 128.
		MaxReadEntries int
		// ReadChunkBytes bounds each filesystem scan buffer. Zero defaults to
		// 256 KiB.
		ReadChunkBytes int
		// MaxConcurrentScans bounds simultaneous scan buffers and descriptors.
		// Zero defaults to 16.
		MaxConcurrentScans int
		// SegmentMaxBytes rotates physical groups before a file exceeds this
		// length. It does not bound one logical record or batch. Zero defaults
		// to 64 MiB.
		SegmentMaxBytes int64
		// SegmentMaxRecords rotates before one segment exceeds this count. Zero
		// defaults to 10,000.
		SegmentMaxRecords int
		// MaxAppendBatchRecords bounds one logical append request by count.
		// Zero defaults to 1,024.
		MaxAppendBatchRecords int
		// MaxAppendBatchBytes bounds one logical request's combined ID and
		// payload bytes across physical groups and segments. Zero defaults to
		// 8 MiB.
		MaxAppendBatchBytes int64
		// MaxCommitGroupRecords bounds one physical commit group by count.
		// Zero defaults to 1,024.
		MaxCommitGroupRecords int
		// MaxCommitGroupBytes bounds one encoded physical commit group and one
		// byte-backed transaction fragment buffer. Zero defaults to 8 MiB.
		MaxCommitGroupBytes int64
		// GroupCommitDelay optionally lets a synchronized leader collect
		// followers before I/O. Zero adds no deliberate delay.
		GroupCommitDelay time.Duration
		// MaxPendingAppendRequests bounds admitted requests waiting for commit.
		// Zero defaults to 4,096.
		MaxPendingAppendRequests int
		// MaxPendingAppendBytes bounds charged pending input. Byte-backed
		// requests and fully prepared reader requests charge actual ID and
		// payload bytes. Zero defaults to 64 MiB.
		MaxPendingAppendBytes int64
		// StagingMemoryThreshold keeps one reader payload in memory through
		// this many bytes before switching to an immutable blob. Zero defaults
		// to 1 MiB.
		StagingMemoryThreshold int
		// MaxStagingRequests bounds concurrent File reader preparations and
		// active reader workers across them. One batch may use several workers.
		// It cannot exceed MaxRecords. Zero defaults to min(16, MaxRecords).
		MaxStagingRequests int
		// MaxStagingBytes bounds their combined worst-case logical byte
		// reservations. Zero defaults to 64 MiB.
		MaxStagingBytes int64
		// MaxIndexBytes bounds retained in-memory record, fragment, segment,
		// transaction-plan, and stable-ID accounting. Zero defaults to 256 MiB.
		MaxIndexBytes int64
		// MetadataReserveBytes keeps bounded space for recovery metadata. Zero
		// defaults to 1 MiB.
		MetadataReserveBytes int64
		// Preallocation controls standby-segment block reservation. Zero selects
		// PreallocateBestEffort.
		Preallocation PreallocationMode
		// OnProgress receives synchronous aggregate operation progress.
		OnProgress Observer
		// OnInternalError receives contained observational callback failures.
		OnInternalError func(error)
		// Logf is the fallback when OnInternalError is absent or panics.
		// Nil or a panic in Logf falls back to log.Printf.
		Logf func(string, ...any)
		// ProgressInterval is the minimum time between periodic progress
		// events. Phase transitions are always reported. Zero defaults to 5s.
		ProgressInterval time.Duration
		faults           *fileFaults
	}

	// FileStats is one lock-consistent segmented WAL snapshot.
	FileStats struct {
		State State
		// LogicalBytes counts retained ID and payload bytes.
		LogicalBytes int64
		// PhysicalBytes counts managed segment file lengths.
		PhysicalBytes int64
		// BlobBytes counts distinct immutable payload file lengths.
		BlobBytes int64
		// StagingBytes counts current staging file lengths.
		StagingBytes int64
		// ManagedBytes counts current managed file lengths plus any active
		// compaction reservation. It excludes active staging reservations.
		ManagedBytes int64
		// ReclaimableBytes estimates acknowledged bytes in the partial head
		// segment that explicit compaction can release. It is zero while the
		// head is a partially retained transaction.
		ReclaimableBytes  int64
		Records           int
		Segments          int
		Blobs             int
		StagingFiles      int
		PreparingRequests int
		PreparingBytes    int64
		// ActiveSegmentBytes is the active segment's committed logical length.
		ActiveSegmentBytes   int64
		ActiveSegmentRecords int
		PendingRequests      int
		// PendingBytes counts charged input for requests awaiting commit.
		PendingBytes  int64
		DirtySegments int
		// SyncedThrough is the highest exact durability watermark established
		// by this instance.
		SyncedThrough Position
		// SyncCount and SyncNanoseconds are saturating aggregate barrier
		// counters.
		SyncCount       uint64
		SyncNanoseconds uint64
		// PreallocationSuccesses and PreallocationFailures count standby
		// reservation outcomes.
		PreallocationSuccesses uint64
		PreallocationFailures  uint64
		ReserveAvailable       bool
		// DataBarrierAvailable and DirectoryBarrierAvailable report compiled
		// primitive availability, not deployment qualification.
		DataBarrierAvailable      bool
		DirectoryBarrierAvailable bool
		// Uncertain reports unresolved append, truncation, or compaction state.
		Uncertain bool
		// SyncFailed reports a terminal synchronization failure.
		SyncFailed bool
	}

	// CompactionReport summarizes one explicit partial-head compaction.
	CompactionReport struct {
		Committed      bool
		BeforeBytes    uint64
		AfterBytes     uint64
		ReclaimedBytes uint64
		ScannedRecords uint64
		CopiedRecords  uint64
		CopiedBytes    uint64
		RemovedFiles   uint64
	}

	// File is one exclusive, segmented filesystem WAL owner.
	File struct {
		log *fileLog
	}
)

// OpenFile opens, validates the requested persisted profile, recovers, and
// returns one segmented WAL. A profile downgrade is rejected before recovery;
// a valid upgrade is prepared and crash-safely persisted before record
// recovery. Cancellation returns no usable owner. Progress observer failures
// are reported internally and do not interrupt configuration or recovery.
func OpenFile(
	ctx context.Context,
	config FileConfig,
) (*File, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}

	log, err := openFileLog(ctx, config)
	if err != nil {
		return nil, err
	}

	return &File{log: log}, nil
}

// Identity returns the immutable identity persisted in the WAL directory.
func (l *File) Identity() Identity {
	if l == nil || l.log == nil || l.log.metadata == nil {
		return Identity{}
	}

	return l.log.metadata.identity.ID
}

// Supports reports whether the current platform can provide durability.
func (l *File) Supports(durability Durability) bool {
	return l != nil &&
		l.log != nil &&
		l.log.supports(durability)
}

// Append commits one record in FIFO position order. Cancellation observed
// before admission returns without mutation. After admission, Append waits for
// and returns the definite storage outcome even when ctx is canceled.
func (l *File) Append(
	ctx context.Context,
	record Record,
	durability Durability,
) (Receipt, error) {
	if l == nil || l.log == nil {
		return Receipt{}, ErrClosed
	}

	return l.log.append(ctx, record, durability)
}

// AppendBatch atomically commits one logical record batch. Cancellation before
// admission rejects the complete batch. Cancellation after admission does not
// interrupt or obscure its definite storage outcome.
func (l *File) AppendBatch(
	ctx context.Context,
	records []Record,
	durability Durability,
) ([]Receipt, error) {
	if l == nil || l.log == nil {
		return nil, ErrClosed
	}

	return l.log.appendBatch(ctx, records, durability)
}

// AppendFrom atomically commits one size-bounded reader payload without
// retaining a large payload in memory. Reader consumption occurs after
// bounded staging admission and before ordinary commit admission. Cancellation
// during staging may consume a reader prefix but commits no record; after
// commit admission it follows Append's definite-result contract. SizeHint is
// optional and never overrides actual-byte limits.
func (l *File) AppendFrom(
	ctx context.Context,
	record ReaderRecord,
	durability Durability,
) (Receipt, error) {
	if l == nil || l.log == nil {
		return Receipt{}, ErrClosed
	}

	return l.log.appendFrom(ctx, record, durability)
}

// AppendBatchFrom atomically commits one ordered batch of size-bounded reader
// payloads. Independent readers may be consumed concurrently after bounded
// staging admission, while commit order remains input order. No record becomes
// visible unless every payload reaches EOF within the actual-byte limits and
// the transaction commit marker succeeds.
func (l *File) AppendBatchFrom(
	ctx context.Context,
	records []ReaderRecord,
	durability Durability,
) ([]Receipt, error) {
	if l == nil || l.log == nil {
		return nil, ErrClosed
	}

	return l.log.appendBatchFrom(ctx, records, durability)
}

// Read returns copied committed records after one retained position.
func (l *File) Read(
	ctx context.Context,
	after Position,
	limit int,
) ([]Entry, error) {
	if l == nil || l.log == nil {
		return nil, ErrClosed
	}

	return l.log.read(ctx, after, limit)
}

// Scan validates complete commit groups before invoking visit for their
// retained records. Entry.Record.Data is borrowed only for the synchronous
// callback and must not be retained or modified after it returns. The callback
// runs without the WAL mutex.
func (l *File) Scan(
	ctx context.Context,
	after Position,
	limit int,
	visit func(Entry) error,
) error {
	if l == nil || l.log == nil {
		return ErrClosed
	}

	return l.log.scan(ctx, after, limit, visit)
}

// ScanReader validates each complete record before exposing its exact size and
// a bounded reader to the synchronous callback. Fragmented records are reread
// rather than materialized. Reader is valid only until the callback returns.
// The callback runs without the WAL mutex.
func (l *File) ScanReader(
	ctx context.Context,
	after Position,
	limit int,
	visit func(ReaderEntry) error,
) error {
	if l == nil || l.log == nil {
		return ErrClosed
	}

	return l.log.scanReader(ctx, after, limit, visit)
}

// OpenReader opens one exact committed record through a bounded, validated,
// one-shot reader. Close releases its scan lease.
func (l *File) OpenReader(
	ctx context.Context,
	position Position,
) (EntryReader, error) {
	if l == nil || l.log == nil {
		return nil, ErrClosed
	}

	return l.log.openReader(ctx, position)
}

// Flush establishes durability through the returned admitted watermark.
// Cancellation is accepted only before the barrier is admitted. A conflicting
// maintenance operation may return ErrBackpressure before admission.
func (l *File) Flush(
	ctx context.Context,
	durability Durability,
) (Position, error) {
	if l == nil || l.log == nil {
		return 0, ErrClosed
	}

	return l.log.flush(ctx, durability)
}

// Truncate durably checkpoints and removes a committed prefix.
// Cancellation before checkpoint publication leaves the logical prefix
// unchanged. Once publication begins, Truncate returns its definite outcome.
func (l *File) Truncate(
	ctx context.Context,
	through Position,
	durability Durability,
) error {
	if l == nil || l.log == nil {
		return ErrClosed
	}

	return l.log.truncate(ctx, through, durability)
}

// Compact rewrites one partial head segment without changing logical records.
func (l *File) Compact(
	ctx context.Context,
) (CompactionReport, error) {
	if l == nil || l.log == nil {
		return CompactionReport{}, ErrClosed
	}

	return l.log.compact(ctx)
}

// State returns the current logical WAL range.
func (l *File) State() State {
	if l == nil || l.log == nil {
		return State{}
	}

	return l.log.state()
}

// Stats returns a lock-consistent snapshot without filesystem I/O.
func (l *File) Stats() FileStats {
	if l == nil || l.log == nil {
		return FileStats{}
	}

	log := l.log
	log.stagingMutex.Lock()
	log.queueMutex.Lock()
	log.mutex.Lock()
	defer func() {
		log.mutex.Unlock()
		log.queueMutex.Unlock()
		log.stagingMutex.Unlock()
	}()

	syncCount, syncNanoseconds := log.metadata.fileSystem.syncStats()
	objectStats := log.objects.stats()

	stats := FileStats{
		State: State{
			Truncated: log.truncated,
			Last:      log.last,
		},
		LogicalBytes:  log.logicalBytes,
		PhysicalBytes: log.segmentBytes.Load(),
		BlobBytes:     objectStats.BlobBytes,
		StagingBytes:  objectStats.StagingBytes,
		ManagedBytes: log.managedWALBytesWithObjects(
			objectStats,
		),
		Records:                   len(log.entries),
		Segments:                  log.segmentCount,
		Blobs:                     objectStats.Blobs,
		StagingFiles:              objectStats.StagingFiles,
		PreparingRequests:         log.preparingRequests,
		PreparingBytes:            log.preparingBytes,
		ActiveSegmentBytes:        log.activeBytes,
		ActiveSegmentRecords:      log.activeRecords,
		PendingRequests:           log.pendingRequests,
		PendingBytes:              log.pendingBytes,
		DirtySegments:             log.dirtySegments,
		SyncedThrough:             log.syncedThrough,
		SyncCount:                 syncCount,
		SyncNanoseconds:           syncNanoseconds,
		PreallocationSuccesses:    log.metadata.preallocationSuccess,
		PreallocationFailures:     log.metadata.preallocationFailure,
		ReserveAvailable:          log.metadata.reserveAvailable,
		DataBarrierAvailable:      supportsWALDataBarrier(),
		DirectoryBarrierAvailable: supportsWALDirectoryBarrier(),
		Uncertain:                 log.stateUncertain || log.truncatePending || log.compactionPending,
		SyncFailed:                log.syncFailed,
	}

	if len(log.entries) != 0 {
		head := log.entries[0]
		if head.TransactionGeneration == 0 &&
			head.GroupOffset > walSegmentHeaderBytes {
			stats.ReclaimableBytes = head.GroupOffset - walSegmentHeaderBytes
		}
	}

	return stats
}

// Close rejects future operations and releases exclusive directory ownership.
// It returns ErrBackpressure, without beginning close, while reader
// preparation, a commit, a read or scan lease, or conflicting maintenance is
// active.
func (l *File) Close() error {
	if l == nil || l.log == nil {
		return nil
	}

	return l.log.close()
}

var _ StreamingLog = (*File)(nil)