transfer

Documentation

Overview

Package transfer implements verified, resumable file-transfer primitives.

Index

• Constants
• func SendFile(ctx context.Context, client *http.Client, baseURL string, root int, ...) (string, error)
• func SendFileWithProgress(ctx context.Context, client *http.Client, baseURL string, root int, ...) (string, error)
• func SendTreeWithProgress(ctx context.Context, client *http.Client, baseURL string, root int, ...) error
• func SourceName(path string) string
• func ValidateRelative(p string) error
• type Job
• type JobSpec
• type Jobs
  • func NewJobs() *Jobs
  • func NewPersistentJobs(path string) (*Jobs, error)
  • func (j *Jobs) Finish(id string, err error)
  • func (j *Jobs) Get(id string) (Job, bool)
  • func (j *Jobs) List() []Job
  • func (j *Jobs) Progress(id string, completed int)
  • func (j *Jobs) Start(total int) Job
  • func (j *Jobs) StartNamed(total int, name string) Job
  • func (j *Jobs) StartWith(spec JobSpec) Job
• type Manifest
  • func BuildManifest(r io.Reader, chunkSize int64) (Manifest, error)
  • func (m Manifest) ChunkLength(index int) (int64, error)
  • func (m Manifest) Verify(index int, body []byte) error
• type Store
  • func NewStore(dir string) (*Store, error)
  • func (s *Store) Finalize(id, destination string) error
  • func (s *Store) Missing(id string) ([]int, error)
  • func (s *Store) Open(id string, m Manifest) error
  • func (s *Store) String() string
  • func (s *Store) Write(id string, index int, body []byte) error
• type Tree
  • func BuildTree(source string) (Tree, error)
• type TreeFile
• type TreeIntent
• type TreeIntentFile
• type TreeStore
  • func NewTreeStore(rootPath string) *TreeStore
  • func (s *TreeStore) Abort(id string) error
  • func (s *TreeStore) Commit(id string) error
  • func (s *TreeStore) Create(destination string, tree Tree) (string, error)
  • func (s *TreeStore) ListIntents() (map[string]TreeIntent, error)
  • func (s *TreeStore) ResolveStagingPath(id, relative string) (string, error)
  • func (s *TreeStore) StagingRoot(id string) string
  • func (s *TreeStore) Status(id string) (TreeIntent, map[string]bool, error)
  • func (s *TreeStore) Sweep(ttl time.Duration, now time.Time) (int, error)
  • func (s *TreeStore) Touch(id string) error

Constants

const DefaultChunkSize int64 = 4 << 20

Functions

func SendFile

func SendFile(ctx context.Context, client *http.Client, baseURL string, root int, destination, source string) (string, error)

SendFile transfers source directly to an Eta peer, resuming only its missing verified chunks. The coordinator need not proxy file bytes.

func SendFileWithProgress

func SendFileWithProgress(ctx context.Context, client *http.Client, baseURL string, root int, destination, source string, progress func(completed, total int)) (string, error)

SendFileWithProgress reports each peer-acknowledged chunk. A nil callback leaves the protocol behavior identical to SendFile.

func SendTreeWithProgress

func SendTreeWithProgress(ctx context.Context, client *http.Client, baseURL string, root int, destination, source string, tree Tree, progress func(completed, total int)) error

SendTreeWithProgress delivers a source directory tree to the peer via the atomic tree-transfer protocol when the peer supports it (/api/transfer-trees). Files are still sent through the existing per-file /api/transfers flow against staging paths so chunk-level verification and per-file resume reuse is unchanged. On commit the receiver performs a single os.Rename of the staging tree to the destination, giving POSIX-atomic destination materialization.

The sender is robust against legacy peers: a 404 from /api/transfer-trees falls back to the prior per-directory create + piecemeal finalize flow so mixed-version fleets degrade gracefully.

func SourceName

func SourceName(path string) string

func ValidateRelative

func ValidateRelative(p string) error

ValidateRelative rejects empty / dot / dot-dot / absolute / escape paths. Walks the components manually so that "a/../b" (which filepath.Clean collapses to "b") is still caught. Also rejects any segment that resolves to the empty string after cleaning. Exported so HTTP handlers can reuse the same rule for user-supplied paths.

Types

type Job

type Job struct {
	ID              string    `json:"id"`
	Name            string    `json:"name,omitempty"`
	Completed       int       `json:"completed"`
	Total           int       `json:"total"`
	Error           string    `json:"error,omitempty"`
	Done            bool      `json:"done"`
	Started         time.Time `json:"started"`
	SourcePeer      string    `json:"sourcePeer,omitempty"`
	SourceRoot      int       `json:"sourceRoot,omitempty"`
	SourcePath      string    `json:"sourcePath,omitempty"`
	DestinationPeer string    `json:"destinationPeer,omitempty"`
	DestinationRoot int       `json:"destinationRoot,omitempty"`
	DestinationPath string    `json:"destinationPath,omitempty"`
}

Job is a durable snapshot of an asynchronous outbound transfer. SourcePeer / SourcePath and DestinationPeer / DestinationPath carry the routing needed for the server to auto-resume the transfer after a process restart. Empty peer means a local endpoint. Jobs whose fields are populated and are not Done are left in that state on restart so the resume goroutine can pick them up; jobs without routing info are too old to resume and are marked interrupted.

type JobSpec

type JobSpec struct {
	Name            string
	Total           int
	SourcePeer      string
	SourceRoot      int
	SourcePath      string
	DestinationPeer string
	DestinationRoot int
	DestinationPath string
}

JobSpec bundles the routing and size hints required to start a Job. It is intentionally separate from Job's runtime fields so that StartWith has no overlap with the polled-progress mutation path.

type Jobs

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

Jobs keeps small control-plane records. The optional path persists snapshots atomically; a process restart marks in-flight work interrupted rather than pretending the sender goroutine can resume by itself.

func NewJobs

func NewJobs() *Jobs

func NewPersistentJobs

func NewPersistentJobs(path string) (*Jobs, error)

func (*Jobs) Finish

func (j *Jobs) Finish(id string, err error)

func (*Jobs) Get

func (j *Jobs) Get(id string) (Job, bool)

func (*Jobs) List

func (j *Jobs) List() []Job

func (*Jobs) Progress

func (j *Jobs) Progress(id string, completed int)

func (*Jobs) Start

func (j *Jobs) Start(total int) Job

func (*Jobs) StartNamed

func (j *Jobs) StartNamed(total int, name string) Job

func (*Jobs) StartWith

func (j *Jobs) StartWith(spec JobSpec) Job

StartWith records a new in-flight Job with the given routing. The returned Job is also persisted; callers Progress/Finish to update it.

type Manifest

type Manifest struct {
	Size      int64    `json:"size"`
	ChunkSize int64    `json:"chunkSize"`
	Chunks    []string `json:"chunks"`
}

func BuildManifest

func BuildManifest(r io.Reader, chunkSize int64) (Manifest, error)

func (Manifest) ChunkLength

func (m Manifest) ChunkLength(index int) (int64, error)

func (Manifest) Verify

func (m Manifest) Verify(index int, body []byte) error

type Store

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

func NewStore

func NewStore(dir string) (*Store, error)

func (*Store) Finalize

func (s *Store) Finalize(id, destination string) error

func (*Store) Missing

func (s *Store) Missing(id string) ([]int, error)

func (*Store) Open

func (s *Store) Open(id string, m Manifest) error

func (*Store) String

func (s *Store) String() string

func (*Store) Write

func (s *Store) Write(id string, index int, body []byte) error

type Tree

type Tree struct {
	Directories []string
	Files       []TreeFile
	TotalChunks int
}

Tree is a deterministic, symlink-free description of a source directory. Paths are slash-separated and relative to the source directory.

func BuildTree

func BuildTree(source string) (Tree, error)

BuildTree validates a directory tree and records the exact file manifests needed for a direct transfer. Symlinks and special files are intentionally excluded rather than silently following or copying them.

type TreeFile

type TreeFile struct {
	Path     string
	Manifest Manifest
}

type TreeIntent

type TreeIntent struct {
	Version      int              `json:"version"`
	ID           string           `json:"id"`
	Destination  string           `json:"destination"`
	Directories  []string         `json:"directories"`
	Files        []TreeIntentFile `json:"files"`
	Created      time.Time        `json:"created"`
	LastProgress time.Time        `json:"lastProgress"`
}

TreeIntent is the durable per-session record. Persisted atomically alongside the staging directory under {root}/.eta/intents/{id}.json.

type TreeIntentFile

type TreeIntentFile struct {
	Path string `json:"path"`
	Size int64  `json:"size"`
}

TreeIntentFile is the per-file record kept in the intent. Size is the pre-transfer byte length so commit can verify presence without keeping full chunk manifests in the intent record (which would balloon for big trees).

type TreeStore

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

TreeStore manages a per-root tree-transfer session lifecycle. A session reserves a destination, stages every file under that root's hidden staging directory, and atomically promotes the staging tree to the destination via a single os.Rename. The promotion is POSIX-atomic on any single filesystem; no half-copied destination tree can ever be observed.

The contract this store enforces:

• destination must NOT exist before Create (single-rename semantics)
• files & directories inside the staging tree must be valid relative paths
• the staging root (".eta") is reserved and never user-creatable
• commit either succeeds completely or leaves the destination absent

Staging lives inside the destination root on purpose: it removes the EXDEV cross-filesystem rename hazard that per-file staging (under the user cache directory) currently has against NAS-mounted roots.

func NewTreeStore

func NewTreeStore(rootPath string) *TreeStore

NewTreeStore binds a tree session manager to a single filesystem root.

func (*TreeStore) Abort

func (s *TreeStore) Abort(id string) error

Abort removes the staging tree and intent record without committing.

func (*TreeStore) Commit

func (s *TreeStore) Commit(id string) error

Commit verifies that every file in the intent is fully written under the staging tree, then performs a single os.Rename of the staging root to the destination. Either the entire destination tree appears or the rename fails and the staging tree remains for retry/abort.

func (*TreeStore) Create

func (s *TreeStore) Create(destination string, tree Tree) (string, error)

Create reserves a destination and creates the staging tree. The supplied Tree describes the directories and files that will be sent. On success returns the new intent ID and persists the intent record. On any error the partially-built staging directory is removed.

func (*TreeStore) ListIntents

func (s *TreeStore) ListIntents() (map[string]TreeIntent, error)

ListIntents returns every persisted tree-intent record on this root, keyed by id. Used by the receiver to discover in-flight sessions during per-file finalize (so it can refresh LastProgress on the matching tree). Returns an empty map when no intents exist.

func (*TreeStore) ResolveStagingPath

func (s *TreeStore) ResolveStagingPath(id, relative string) (string, error)

ResolveStagingPath returns the absolute staging path a sender must target for a given file (relative to the session's destination tree).

func (*TreeStore) StagingRoot

func (s *TreeStore) StagingRoot(id string) string

StagingRoot returns the per-session staging directory for the given ID. Useful for tests; the HTTP API resolves this internally.

func (*TreeStore) Status

func (s *TreeStore) Status(id string) (TreeIntent, map[string]bool, error)

Status returns the current intent plus a per-file "complete" map derived from the filesystem: each file is complete iff a regular file exists at its staging path with the expected size. Cheap (no chunk manifest in intent) and survives crashes.

func (*TreeStore) Sweep

func (s *TreeStore) Sweep(ttl time.Duration, now time.Time) (int, error)

Sweep removes intent records and their staging trees whose LastProgress is older than ttl relative to now. Orphan staging dirs (no matching intent record) are also removed. Returns the number of sessions swept.

func (*TreeStore) Touch

func (s *TreeStore) Touch(id string) error

Touch refreshes the intent's LastProgress timestamp. Callers should invoke this after each chunk PUT or per-file finalize so crash-recovery sweeps can distinguish in-flight transfers from abandoned ones.