Documentation
¶
Overview ¶
Package transfer is a small in-memory tracker for long-running file move/copy operations (post-download move, Local-tab move, AI/promote), so the UI can show ONE consistent progress pattern for all of them: X/Y files, bytes done/total, transfer rate and ETA. It is the move-side analogue of internal/localstream (which meters streaming reads) and reuses the same windowed-rate idea. State is in-memory and lost on restart — for durable resume of the post-download move, see downloads.StatusMoving / RescueStuckMoving.
Index ¶
- func ProgressReader(r io.Reader, onBytes func(int64)) io.Reader
- func ProgressReaderCtx(ctx context.Context, r io.Reader, onBytes func(int64)) io.Reader
- type Job
- func (j *Job) AddBytes(n int64)
- func (j *Job) AddBytesFunc() func(int64)
- func (j *Job) AddSkipped(n int64)
- func (j *Job) Canceled() bool
- func (j *Job) Context() context.Context
- func (j *Job) Done()
- func (j *Job) Fail(err error)
- func (j *Job) FileDone()
- func (j *Job) ID() string
- func (j *Job) SetBytesTotal(total int64)
- func (j *Job) Snapshot() Snapshot
- type Pending
- type Snapshot
- type Status
- type Store
- type Tracker
- func (t *Tracker) ActiveCount() int
- func (t *Tracker) Cancel(id string, userID int, includeAll bool) bool
- func (t *Tracker) List(userID int, includeAll bool) []Snapshot
- func (t *Tracker) Start(label, kind string, filesTotal int, bytesTotal int64) *Job
- func (t *Tracker) StartFor(userID int, label, kind string, filesTotal int, bytesTotal int64) *Job
- func (t *Tracker) Submit(label, kind string, filesTotal int, bytesTotal int64, fn func(*Job)) *Job
- func (t *Tracker) SubmitFor(userID int, label, kind string, filesTotal int, bytesTotal int64, ...) *Job
- func (t *Tracker) WaitIdle(ctx context.Context) bool
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ProgressReader ¶
ProgressReader wraps r so each Read reports the byte count via onBytes — drop-in for io.Copy(dst, ProgressReader(src, job.AddBytes)).
func ProgressReaderCtx ¶
ProgressReaderCtx is ProgressReader that also aborts the copy when ctx is canceled (returns ctx.Err() from Read), so a Tracker.Cancel stops an in-progress file copy mid-stream. nil ctx → no cancellation (plain progress).
Types ¶
type Job ¶
type Job struct {
// contains filtered or unexported fields
}
Job is one tracked move/copy operation. Safe for concurrent use: producers call AddBytes/FileDone from the copy loop while the API reads Snapshot.
func (*Job) AddBytes ¶
AddBytes records bytes transferred (drives progress + rate). No-op on <=0 or a nil Job (so producers can stay agnostic to whether tracking is wired).
func (*Job) AddBytesFunc ¶
AddBytesFunc returns a callback for instrumenting an io.Copy via ProgressReader.
func (*Job) AddSkipped ¶
AddSkipped advances progress by bytes that were NOT copied now because they already existed at the destination (a resumed transfer skipping a file that a previous run finished). Counts toward done/total/progress but does NOT enter the rate window — otherwise skipping a large file instantly would spike the reported speed to an absurd value.
func (*Job) Context ¶
Context returns the job's cancellation context (Background for a nil job). A producer should pass it to its copy loop / retries so Tracker.Cancel aborts the work in flight.
func (*Job) Done ¶
func (j *Job) Done()
Done marks the job successful. Fail marks it failed with the error message.
func (*Job) FileDone ¶
func (j *Job) FileDone()
FileDone increments the completed-files counter (X of Y).
func (*Job) ID ¶
ID returns the job's stable identifier (empty for a nil Job), so a producer can hand it back to the client to correlate with the dock entry.
func (*Job) SetBytesTotal ¶
SetBytesTotal sets/raises the known total byte size (when discovered late).
type Pending ¶
type Pending struct {
ID int64
Kind string // "promote" | "local-move" | "local-promote"
Src string
Dst string
Payload string // kind-specific JSON (promote: downloadID/userID/keepSeeding)
}
Pending is a transfer (move/promote) whose copy must survive a restart. The intent is persisted BEFORE the copy starts and removed when it completes, so a boot reconciler can re-submit whatever was interrupted by a deploy/crash. The copy itself is resume-aware (copyFileAndRemove skips files already at the destination), so a re-submit only finishes what's left — it never re-copies what a previous run already moved.
type Snapshot ¶
type Snapshot struct {
ID string `json:"id"`
Label string `json:"label"`
Kind string `json:"kind"`
UserID int `json:"userId,omitempty"`
Status Status `json:"status"`
FilesDone int `json:"filesDone"`
FilesTotal int `json:"filesTotal"`
BytesDone int64 `json:"bytesDone"`
BytesTotal int64 `json:"bytesTotal"`
RatePerSec int64 `json:"ratePerSec"`
ETASeconds int `json:"etaSeconds"`
Progress float64 `json:"progress"` // 0..1 (by bytes when known, else by files)
Error string `json:"error,omitempty"`
StartedAt string `json:"startedAt"`
}
Snapshot is the immutable, JSON-serializable view of a Job for the UI.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store persists pending transfers in PostgreSQL (shared pool). nil-safe so callers can stay agnostic to whether persistence is wired.
func OpenStore ¶
OpenStore wires the pending-transfers store onto the shared Postgres pool. Schema is applied centrally (internal/db migrations).
func (*Store) Close ¶
func (s *Store) Close()
Close is a no-op: the shared pool's lifecycle is owned by main.
type Tracker ¶
type Tracker struct {
// contains filtered or unexported fields
}
Tracker holds the active and recently-finished jobs.
func New ¶
New returns a Tracker using the wall clock. maxConcurrent (optional, default 3) caps simultaneous Submit() transfers; excess ones queue.
func (*Tracker) ActiveCount ¶
ActiveCount returns how many jobs are still Queued or Running. Used by the graceful shutdown to decide whether to wait for in-flight moves.
func (*Tracker) Cancel ¶
Cancel cancels a tracked job by ID. When includeAll is false, only a job owned by userID may be canceled (userID 0 jobs are system-owned and cancelable by anyone).
func (*Tracker) List ¶
List returns snapshots of jobs (newest first). When includeAll is false, only jobs owned by userID (plus system jobs with userID 0) are included.
func (*Tracker) Start ¶
Start registers and returns a new RUNNING Job (no queueing). filesTotal/ bytesTotal may be 0 when unknown (rate/ETA degrade gracefully). userID=0 is anonymous/system; prefer StartFor when the owner is known.
func (*Tracker) StartFor ¶ added in v0.97.0
StartFor is Start with an explicit owner userID for multi-tenant filtering.
func (*Tracker) Submit ¶
Submit registers a job that starts QUEUED and runs fn once a concurrency slot frees (bounded by maxConcurrent; excess jobs wait FIFO). fn receives the now- running Job and owns its terminal Done()/Fail(). Returns immediately. On a nil Tracker, fn runs unbounded in a goroutine with a nil Job (tracking disabled).
func (*Tracker) SubmitFor ¶ added in v0.97.0
func (t *Tracker) SubmitFor(userID int, label, kind string, filesTotal int, bytesTotal int64, fn func(*Job)) *Job
SubmitFor is Submit with an explicit owner userID.
Source Files