multipart

Documentation

Index

• Variables
• type InitReq
• type Manager
  • func (m *Manager) Abort(user *model.User, id string) error
  • func (m *Manager) Chunk(user *model.User, id string, idx int, body io.Reader) (SessionSnapshot, error)
  • func (m *Manager) Complete(ctx context.Context, user *model.User, id string) (SessionSnapshot, error)
  • func (m *Manager) Find(user *model.User, path string, size int64) (SessionSnapshot, error)
  • func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error)
  • func (m *Manager) StartGC()
  • func (m *Manager) Status(user *model.User, id string) (SessionSnapshot, error)
• type Session
  • func (s *Session) Snapshot() SessionSnapshot
• type SessionSnapshot
• type Snapshot
• type State
• type Window
  • func NewWindow(dir, id string, chunkSize, totalSize int64, slots int) (*Window, error)
  • func (w *Window) CRCs() ([]uint32, []bool)
  • func (w *Window) ChunkLen(idx int) int64
  • func (w *Window) Close() error
  • func (w *Window) CloseWithError(sticky error) error
  • func (w *Window) Read(p []byte) (int, error)
  • func (w *Window) Snapshot() Snapshot
  • func (w *Window) TotalChunks() int
  • func (w *Window) WriteChunk(idx int, r io.Reader) (uint32, error)

Variables

var (
	ErrSessionNotFound = errors.New("multipart upload session not found")
	ErrNotOwner        = errors.New("multipart upload session belongs to another user")
)

var (
	// ErrClosed is the sticky error after Close; all pending and future reads/writes fail with it.
	ErrClosed = errors.New("multipart upload window closed")
	// ErrChunkInFlight means another request is uploading the same chunk right now.
	ErrChunkInFlight = errors.New("chunk is being uploaded by another request")
	// ErrOutOfWindow means the chunk is still too far ahead of the consumption
	// frontier after waiting WindowWaitTimeout; the client should back off and
	// resend it later (flow control, not a failure).
	ErrOutOfWindow = errors.New("chunk is out of the receiving window")
)

var DefaultManager = &Manager{
	byID:  make(map[string]*Session),
	byKey: make(map[string]string),
}

var WindowSlots = 8

WindowSlots bounds the per-session disk footprint to WindowSlots*ChunkSize.

var WindowWaitTimeout = 10 * time.Second

WindowWaitTimeout bounds how long WriteChunk blocks waiting for its slot. Browsers cannot reliably read responses sent before the request body is consumed (they report a network error), so under backpressure it is far better to hold the request until a slot frees — the wait must just stay well below CDN request deadlines (Cloudflare: ~100s). Tests shrink this.

Types

type InitReq

type InitReq struct {
	User      *model.User
	Path      string // full destination path, already user-joined
	Size      int64
	ChunkSize int64 // final chunk size in bytes, already clamped by the handler
	Mimetype  string
	Modified  time.Time
	Hashes    map[*utils.HashType]string
}

InitReq carries everything the handler parsed from the init request.

type Manager

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

Manager owns all live sessions. Sessions are in-memory only (aligned with upload tasks not being persisted); a restart drops them and the ring files are swept on the next start.

func (*Manager) Abort

func (m *Manager) Abort(user *model.User, id string) error

Abort cancels the pipeline and forgets the session immediately.

func (*Manager) Chunk

func (m *Manager) Chunk(user *model.User, id string, idx int, body io.Reader) (SessionSnapshot, error)

Chunk feeds one chunk into the session. Re-sending chunk 0 to a failed_retriable session re-fills it: a fresh window and pipeline attempt.

func (*Manager) Complete

func (m *Manager) Complete(ctx context.Context, user *model.User, id string) (SessionSnapshot, error)

Complete waits for the pipeline outcome. It refuses to block while chunks are still missing, so a buggy client cannot park a connection for the TTL.

func (*Manager) Find

func (m *Manager) Find(user *model.User, path string, size int64) (SessionSnapshot, error)

Find looks a live session up by destination path and size, for resume discovery.

func (*Manager) Init

func (m *Manager) Init(req InitReq) (SessionSnapshot, bool, error)

Init creates a session and starts its pipeline, or returns the live session for the same (user, path, size) so an interrupted client resumes implicitly.

func (*Manager) StartGC

func (m *Manager) StartGC()

StartGC sweeps ring files orphaned by a previous run and starts the expiry loop. It is called at server startup so orphans are reclaimed even if no multipart upload ever happens again; Init also calls it, so embedders that skip the server wiring still get GC lazily.

func (*Manager) Status

func (m *Manager) Status(user *model.User, id string) (SessionSnapshot, error)

Status looks a session up by id.

type Session

type Session struct {
	ID        string
	Path      string // full destination path (dir + name), already user-joined
	DstDir    string
	Name      string
	Size      int64
	ChunkSize int64
	Total     int
	Mimetype  string
	Modified  time.Time
	Hashes    map[*utils.HashType]string
	Creator   *model.User
	// contains filtered or unexported fields
}

Session is one multipart upload: metadata survives pipeline attempts, the window (chunk data) does not.

func (*Session) Snapshot

func (s *Session) Snapshot() SessionSnapshot

type SessionSnapshot

type SessionSnapshot struct {
	ID              string   `json:"upload_id"`
	State           State    `json:"state"`
	Attempt         int      `json:"attempt"`
	Path            string   `json:"path"`
	Size            int64    `json:"size"`
	ChunkSize       int64    `json:"chunk_size"`
	TotalChunks     int      `json:"total_chunks"`
	Received        [][2]int `json:"received"`
	ReceivedBytes   int64    `json:"received_bytes"`
	Frontier        int      `json:"frontier"`
	StorageProgress float64  `json:"storage_progress"`
	Error           string   `json:"error,omitempty"`
}

Snapshot is the wire representation of a session used by all endpoints.

type Snapshot

type Snapshot struct {
	// Frontier is the next chunk index to be consumed (== TotalChunks when the stream is fully consumed).
	Frontier int
	// ReadPos is the number of bytes already consumed by the pipeline.
	ReadPos int64
	// ReceivedBytes is the number of payload bytes received from the client (consumed + buffered).
	ReceivedBytes int64
	// Received holds inclusive ranges of chunk indexes the client does not need to resend.
	Received [][2]int
}

Snapshot describes the receiving state, used for status responses and resume.

type State

type State string

const (
	StateReceiving       State = "receiving"
	StateCompleted       State = "completed"
	StateFailedRetriable State = "failed_retriable"
	StateFailedPermanent State = "failed_permanent"
	StateAborted         State = "aborted"
)

type Window

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

Window reassembles concurrently uploaded chunks into a sequential stream. Chunks land in a ring file of slots*chunkSize bytes (chunk i -> slot i%slots), and Read serves bytes in order, blocking until the next needed chunk arrives. A chunk slot is released as soon as the reader crosses its boundary, so the disk footprint is bounded by slots*chunkSize regardless of the file size.

WriteChunk is safe for concurrent use; Read must be called from a single goroutine (the same contract as the FileStreamer it backs).

func NewWindow

func NewWindow(dir, id string, chunkSize, totalSize int64, slots int) (*Window, error)

func (*Window) CRCs

func (w *Window) CRCs() ([]uint32, []bool)

CRCs returns a copy of the per-chunk CRC32 table and which entries are set. It remains readable after Close, so the session can compare re-filled chunks against a previous attempt.

func (*Window) ChunkLen

func (w *Window) ChunkLen(idx int) int64

ChunkLen returns the payload length of chunk idx (the last chunk may be short).

func (*Window) Close

func (w *Window) Close() error

Close makes all pending and future operations fail with ErrClosed and removes the ring file. It is invoked by op.Put via FileStream.Closers when the pipeline ends, and is safe to call multiple times.

func (*Window) CloseWithError

func (w *Window) CloseWithError(sticky error) error

CloseWithError is Close with a caller-chosen sticky error. The session's abort path passes an error wrapping context.Canceled so that a driver woken up from a blocked Read treats the abort exactly like a canceled request (e.g. the local driver only removes partially written files in that case).

func (*Window) Read

func (w *Window) Read(p []byte) (int, error)

Read serves the reassembled stream in order, blocking until the next chunk is available, the window is closed, or the stream ends (io.EOF).

func (*Window) Snapshot

func (w *Window) Snapshot() Snapshot

Snapshot reports the receiving state for status responses and resume discovery.

func (*Window) TotalChunks

func (w *Window) TotalChunks() int

func (*Window) WriteChunk

func (w *Window) WriteChunk(idx int, r io.Reader) (uint32, error)

WriteChunk reads exactly the chunk payload from r into the ring and returns its CRC32 (IEEE). Re-sending an already buffered or consumed chunk succeeds immediately without touching data.