transfer

package
v0.1.11 Latest Latest
Warning

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

 Go to latest Published: Aug 21, 2026 License: GPL-3.0 Imports: 18 Imported by: 0

Documentation

Overview

Package transfer provides robust file transfer implementations with timeout, retry, progress tracking, and checksum verification. It solves the problem of os.Rename() and io.Copy() hanging indefinitely on failing disks.

The primary implementation uses rsync as a backend, which has built-in timeout handling (--timeout flag) that aborts transfers when no progress is made for a specified duration.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrTimeout is returned when a transfer times out due to no progress
	ErrTimeout = errors.New("transfer timed out: no progress")

	// ErrChecksumMismatch is returned when post-transfer checksum verification fails
	ErrChecksumMismatch = errors.New("checksum mismatch after transfer")

	// ErrSourceNotFound is returned when the source file doesn't exist
	ErrSourceNotFound = errors.New("source file not found")

	// ErrDestinationNotWritable is returned when the destination is not writable
	ErrDestinationNotWritable = errors.New("destination not writable")

	// ErrDiskUnhealthy is returned when pre-flight disk health check fails
	ErrDiskUnhealthy = errors.New("disk health check failed")

	// ErrTransferFailed is returned when a transfer fails for unspecified reasons
	ErrTransferFailed = errors.New("transfer failed")

	// ErrRetryExhausted is returned when all retry attempts have been exhausted
	ErrRetryExhausted = errors.New("all retry attempts exhausted")
)

Common errors returned by transfer operations

Functions

func ApplyPermissions 

func ApplyPermissions(path string, opts TransferOptions) error

ApplyPermissions sets file mode and ownership based on TransferOptions. Requires root privileges for chown operations.

func CheckDiskHealthForTransfer 

func CheckDiskHealthForTransfer(src, dst string, timeout time.Duration, requiredSpace int64) error

CheckDiskHealthForTransfer verifies source (if non-empty) and destination directories are healthy enough to begin a transfer. The `timeout` parameter applies to stat/statfs; the write probe uses max(timeout, 30s) so concurrent transfers to the same disk don't trigger a false-positive failure.

func OpenWithTimeout 

func OpenWithTimeout(path string, flag int, perm os.FileMode, timeout time.Duration) (*os.File, error)

func RemoveWithTimeout 

func RemoveWithTimeout(path string, timeout time.Duration) error

func StatWithTimeout 

func StatWithTimeout(path string, timeout time.Duration) (os.FileInfo, error)

Types

type Backend 

type Backend int
const (
	BackendAuto Backend = iota
	BackendPV
	BackendRsync
	BackendNative
)

func ParseBackend 

func ParseBackend(s string) Backend

func (Backend) String 

func (b Backend) String() string

type DiskHealth 

type DiskHealth struct {
	Path       string
	Accessible bool
	Writable   bool
	SpaceFree  int64
	SpaceTotal int64
	IOErrors   bool
	MountOK    bool
	Error      error
}

func CheckDiskHealth 

func CheckDiskHealth(path string, timeout time.Duration) (*DiskHealth, error)

CheckDiskHealth probes a directory with separate timeouts for stat, statfs, and a write probe. The legacy single-timeout signature uses `timeout` for stat and statfs and a more generous budget (max(timeout, 30s)) for the write probe so that brief I/O contention from concurrent transfers does not flag a healthy disk as unhealthy.

func CheckDiskHealthDetailed 

func CheckDiskHealthDetailed(path string, statTimeout, statfsTimeout, writeTimeout time.Duration) (*DiskHealth, error)

CheckDiskHealthDetailed runs each probe under its own timeout. Use this when the destination disk may be under heavy concurrent write load: the write probe contends with active streams and needs a larger budget than stat/statfs.

func (*DiskHealth) IsHealthy 

func (h *DiskHealth) IsHealthy() bool

type FallbackTransferer 

type FallbackTransferer struct {
	// contains filtered or unexported fields
}

FallbackTransferer tries multiple backends in order until one succeeds.

func NewFallbackTransferer 

func NewFallbackTransferer(backends ...Transferer) *FallbackTransferer

NewFallbackTransferer creates a transferer that tries backends in order. If the first backend fails, it tries the next, and so on.

func (*FallbackTransferer) CanResume 

func (f *FallbackTransferer) CanResume() bool

func (*FallbackTransferer) Copy 

func (f *FallbackTransferer) Copy(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*FallbackTransferer) Move 

func (f *FallbackTransferer) Move(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*FallbackTransferer) Name 

func (f *FallbackTransferer) Name() string

type NativeTransferer 

type NativeTransferer struct {
	// contains filtered or unexported fields
}

func NewNativeTransferer 

func NewNativeTransferer(bufferSize int) *NativeTransferer

func (*NativeTransferer) CanResume 

func (n *NativeTransferer) CanResume() bool

func (*NativeTransferer) Copy 

func (n *NativeTransferer) Copy(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*NativeTransferer) Move 

func (n *NativeTransferer) Move(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*NativeTransferer) Name 

func (n *NativeTransferer) Name() string

type PVTransferer 

type PVTransferer struct {
	// contains filtered or unexported fields
}

func NewPVTransferer 

func NewPVTransferer(pvPath string) *PVTransferer

func (*PVTransferer) CanResume 

func (p *PVTransferer) CanResume() bool

func (*PVTransferer) Copy 

func (p *PVTransferer) Copy(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*PVTransferer) Move 

func (p *PVTransferer) Move(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*PVTransferer) Name 

func (p *PVTransferer) Name() string

type RsyncTransferer 

type RsyncTransferer struct {
	// contains filtered or unexported fields
}

func NewRsyncTransferer 

func NewRsyncTransferer(rsyncPath string) *RsyncTransferer

func (*RsyncTransferer) CanResume 

func (r *RsyncTransferer) CanResume() bool

func (*RsyncTransferer) Copy 

func (r *RsyncTransferer) Copy(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*RsyncTransferer) Move 

func (r *RsyncTransferer) Move(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*RsyncTransferer) Name 

func (r *RsyncTransferer) Name() string

type TransferOptions 

type TransferOptions struct {
	// Timeout specifies how long to wait without progress before aborting.
	// A value of 0 means no timeout.
	Timeout time.Duration

	// Checksum enables post-transfer checksum verification.
	Checksum bool

	// Progress is called periodically with transfer progress updates.
	// current is bytes transferred so far, total is total bytes to transfer.
	// total may be -1 if unknown.
	Progress func(current, total int64)

	// RetryAttempts specifies how many times to retry on transient failures.
	// A value of 0 means no retries.
	RetryAttempts int

	// RetryDelay specifies how long to wait between retry attempts.
	RetryDelay time.Duration

	// PreserveAttrs preserves file ownership and permissions during transfer.
	PreserveAttrs bool

	// PreserveTimes preserves source modification times during transfer.
	// Default false: organized files get current mtimes so Jellyfin's
	// DateCreated (Recently Added) reflects when the file entered the library.
	PreserveTimes bool

	// DeletePartial removes partial files if transfer fails.
	// If false, partial files are left for potential resumption.
	DeletePartial bool

	// TargetUID sets the owner of transferred files. A value of -1 means
	// preserve source ownership (or process owner if PreserveAttrs is false).
	TargetUID int

	// TargetGID sets the group of transferred files. A value of -1 means
	// preserve source group (or process group if PreserveAttrs is false).
	TargetGID int

	// FileMode sets the permissions for transferred files. A value of 0 means
	// preserve source permissions (or use umask default if PreserveAttrs is false).
	FileMode os.FileMode

	// DirMode sets the permissions for created directories. A value of 0 means
	// use 0755 default.
	DirMode os.FileMode

	// SkipHealthCheck bypasses the per-backend pre-flight CheckDiskHealthForTransfer.
	// Set by orchestrators (e.g. FallbackTransferer) that have already verified
	// disk health to avoid redundant write probes that contend with active streams.
	SkipHealthCheck bool
}

TransferOptions configures the behavior of a file transfer operation.

func DefaultOptions 

func DefaultOptions() TransferOptions

DefaultOptions returns sensible default transfer options.

func OptionsFromConfig 

func OptionsFromConfig(cfg *config.Config) TransferOptions

OptionsFromConfig creates TransferOptions from configuration. It starts with DefaultOptions() and applies any permission settings from cfg. If cfg is nil, it returns the default options unchanged.

type TransferResult 

type TransferResult struct {
	// Success indicates whether the transfer completed successfully
	Success bool

	// BytesTotal is the total size of the source file
	BytesTotal int64

	// BytesCopied is the number of bytes actually transferred
	BytesCopied int64

	// Duration is how long the transfer took
	Duration time.Duration

	// Checksum is the checksum of the transferred file (if verification was enabled)
	Checksum string

	// SourceRemoved indicates whether the source file was deleted (for Move operations)
	SourceRemoved bool

	// Attempts is the number of attempts made (including retries)
	Attempts int

	// Error contains the error if Success is false
	Error error
}

TransferResult contains details about a completed transfer operation.

type Transferer 

type Transferer interface {
	// Move transfers a file from src to dst, then removes the source.
	// The source is only removed after successful transfer verification.
	// Returns ErrSourceNotFound if source doesn't exist.
	// Returns ErrDestinationNotWritable if destination isn't writable.
	// Returns ErrTimeout if no progress is made within opts.Timeout.
	Move(src, dst string, opts TransferOptions) (*TransferResult, error)

	// Copy transfers a file from src to dst without removing the source.
	// Same error conditions as Move.
	Copy(src, dst string, opts TransferOptions) (*TransferResult, error)

	// CanResume returns true if this transferer supports resuming
	// interrupted transfers.
	CanResume() bool

	// Name returns a human-readable name for this transferer implementation.
	Name() string
}

Transferer is the interface for file transfer implementations. Implementations must be safe for concurrent use.

func MustNew 

func MustNew(backend Backend) Transferer

MustNew creates a new Transferer with the specified backend. Deprecated: Use New instead. MustNew panics on error which is not recommended for production code. Migrate to New(backend) (Transferer, error) for proper error handling.

func New 

func New(backend Backend) (Transferer, error)

type VolumeLimitedTransferer 

type VolumeLimitedTransferer struct {
	// contains filtered or unexported fields
}

VolumeLimitedTransferer wraps a Transferer and enforces a per-destination volume concurrency cap on Move/Copy operations. The underlying transferer is invoked unchanged once a slot is acquired.

func NewVolumeLimitedTransferer 

func NewVolumeLimitedTransferer(inner Transferer, limiter *VolumeLimiter) *VolumeLimitedTransferer

NewVolumeLimitedTransferer wraps `inner` with the supplied limiter. If limiter is nil or its cap is <= 0, this is effectively a passthrough.

func (*VolumeLimitedTransferer) CanResume 

func (t *VolumeLimitedTransferer) CanResume() bool

func (*VolumeLimitedTransferer) Copy 

func (t *VolumeLimitedTransferer) Copy(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*VolumeLimitedTransferer) Move 

func (t *VolumeLimitedTransferer) Move(src, dst string, opts TransferOptions) (*TransferResult, error)

func (*VolumeLimitedTransferer) Name 

func (t *VolumeLimitedTransferer) Name() string

type VolumeLimiter 

type VolumeLimiter struct {
	// contains filtered or unexported fields
}

VolumeLimiter caps the number of concurrent transfers per destination volume (mount point). Heavy concurrent rsync to the same disk causes I/O contention that triggered the false-positive "destination disk unhealthy" cascade in the original health check, and produces long no-progress timeouts on legitimately large transfers.

The limiter detects the mount root by walking up parent directories until the device number changes (standard Linux mount detection). Each mount root gets its own buffered semaphore.

func NewVolumeLimiter 

func NewVolumeLimiter(concurrent int) *VolumeLimiter

NewVolumeLimiter returns a limiter that allows up to `concurrent` parallel transfers per destination mount point. A value <= 0 disables limiting.

func (*VolumeLimiter) Acquire 

func (v *VolumeLimiter) Acquire(dst string) func()

Acquire blocks until a slot is available for the volume containing dst, then returns a release function. Callers must invoke release exactly once (typically `defer release()`). When the limiter is disabled, release is a no-op and Acquire returns immediately.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.