queue

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 6, 2026 License: MPL-2.0 Imports: 12 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Conflict

type Conflict struct {
	Path          string
	LocalVersion  *FileVersion
	RemoteVersion *FileVersion
	DetectedAt    time.Time
}

Conflict represents a detected file conflict.

type ConflictResolution

type ConflictResolution int

ConflictResolution defines how to resolve a file conflict.

const (
	KeepBoth      ConflictResolution = iota // Default: rename local as .conflict.TIMESTAMP
	LastWriteWins                           // Higher mtime wins
	AskUser                                 // UI prompt for choice
)

type ConflictResolver

type ConflictResolver struct {
	Strategy ConflictResolution
	BaseDir  string // Local filesystem base directory
}

ConflictResolver handles file conflicts when both peers modify the same file.

func NewConflictResolver

func NewConflictResolver(baseDir string) *ConflictResolver

NewConflictResolver creates a resolver with KeepBoth as default strategy.

func (*ConflictResolver) CleanupConflict

func (r *ConflictResolver) CleanupConflict(conflictPath string) error

CleanupConflict removes a conflict file after the user resolves it.

func (*ConflictResolver) DetectConflict

func (r *ConflictResolver) DetectConflict(path string, localVer, remoteVer *FileVersion) *Conflict

DetectConflict reports a conflict between the local and remote versions. It returns nil when either version is nil or the checksums match.

func (*ConflictResolver) GetLocalVersion

func (r *ConflictResolver) GetLocalVersion(path string) (*FileVersion, error)

GetLocalVersion reads the current local file version info.

func (*ConflictResolver) ListConflicts

func (r *ConflictResolver) ListConflicts() ([]string, error)

ListConflicts returns all .conflict files in the base directory.

func (*ConflictResolver) Resolve

func (r *ConflictResolver) Resolve(conflict *Conflict) (conflictPath string, err error)

Resolve applies the configured strategy to a conflict. It returns the path where the local version moved (KeepBoth), or an empty string.

type FileVersion

type FileVersion struct {
	Path       string
	Size       int64
	Mtime      int64  // unix nano
	Checksum   uint64 // xxHash3
	SourcePeer string // "local" or peer fingerprint prefix
}

FileVersion tracks file state for conflict detection.

type OperationType

type OperationType int

OperationType defines the type of filesystem operation being queued.

const (
	OpWrite OperationType = iota
	OpCreate
	OpDelete
	OpRename
	OpMkdir
	OpRmdir
)

func (OperationType) String

func (t OperationType) String() string

String returns the operation type as a string.

type QueuedOperation

type QueuedOperation struct {
	ID        uint64 // Monotonic operation ID
	Type      OperationType
	Path      string    // Relative path in the mounted filesystem
	OldPath   string    // Source path for RENAME
	Data      []byte    // Data to write, for WRITE
	Offset    int64     // File offset, for WRITE
	Size      int64     // File size, for CREATE/EDIT
	Mode      uint32    // File mode and permissions
	Mtime     int64     // Modification time (unix nano)
	Checksum  uint64    // xxHash3 of Data, for integrity checks
	CreatedAt time.Time // Queue time
	Retries   int       // Replay attempt count
}

QueuedOperation is a pending filesystem operation to send to the peer.

type ReplayFunc

type ReplayFunc func(op *QueuedOperation) error

ReplayFunc replays one queued operation during Flush.

type WriteQueue

type WriteQueue struct {

	// Configuration
	MaxOps   int   // Maximum queued operations (default 1000)
	MaxBytes int64 // Maximum queued data bytes (default 100 MB)
	// contains filtered or unexported fields
}

WriteQueue persists pending filesystem operations so they survive crashes. While a peer is disconnected, each operation queues as one {id}.json file and replays on reconnect.

func NewWriteQueue

func NewWriteQueue(dir string, sessionID string, logger *slog.Logger) (*WriteQueue, error)

NewWriteQueue creates a write queue that persists operations as JSON files in dir.

func (*WriteQueue) Clear

func (q *WriteQueue) Clear() error

Clear removes all queued operations and their files.

func (*WriteQueue) Close

func (q *WriteQueue) Close() error

Close is a no-op; it exists for interface compatibility.

func (*WriteQueue) Count

func (q *WriteQueue) Count() int

Count returns the number of pending operations.

func (*WriteQueue) DisableQueueing

func (q *WriteQueue) DisableQueueing()

DisableQueueing disables operation queueing. Call it when connected.

func (*WriteQueue) EnableQueueing

func (q *WriteQueue) EnableQueueing()

EnableQueueing enables operation queueing. Call it on disconnect.

func (*WriteQueue) Enqueue

func (q *WriteQueue) Enqueue(op *QueuedOperation) error

Enqueue adds an operation to the queue. It returns nil when queueing is off; the caller then sends the operation directly.

func (*WriteQueue) EnqueueCreate

func (q *WriteQueue) EnqueueCreate(path string, size int64, mode uint32, mtime int64) error

EnqueueCreate is a convenience method for file creation.

func (*WriteQueue) EnqueueDelete

func (q *WriteQueue) EnqueueDelete(path string) error

EnqueueDelete is a convenience method for file deletion.

func (*WriteQueue) EnqueueRename

func (q *WriteQueue) EnqueueRename(oldPath, newPath string) error

EnqueueRename is a convenience method for file rename.

func (*WriteQueue) EnqueueWrite

func (q *WriteQueue) EnqueueWrite(path string, data []byte, offset int64) error

EnqueueWrite is a convenience method for write operations.

func (*WriteQueue) Flush

func (q *WriteQueue) Flush(replay ReplayFunc) []error

Flush replays all queued operations and removes each success from the queue. A failed operation retries up to 3 times, then leaves the queue as a conflict.

func (*WriteQueue) IsEnabled

func (q *WriteQueue) IsEnabled() bool

IsEnabled reports whether queueing is enabled.

func (*WriteQueue) Stats

func (q *WriteQueue) Stats() (enqueued, replayed, failed uint64)

Stats returns queue statistics.

func (*WriteQueue) TotalBytes

func (q *WriteQueue) TotalBytes() int64

TotalBytes returns the total size of queued data.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL