Documentation
¶
Overview ¶
Package staging is filex's own upload staging area: the bytes of an upload land here first, on local disk, and are transferred to the storage driver afterwards by a background op.
Why this exists: before it, the only chunked upload path required storage.MultipartUploader — implemented by the S3 driver alone — and on that path the browser PUT its parts straight to S3, so filex never saw the bytes. Every other write was one synchronous request in which the client waited for the backend, which on a slow backend means the progress bar shows the *backend's* speed and a 4 GB upload over a flaky link starts from zero.
Layout ¶
<root>/<id>/manifest.json <root>/<id>/000001.part <root>/<id>/000002.part …
Numbered parts, deliberately — NOT a single append-only <id>.part with a byte offset. A single file plus an offset serves a sequential resumable client and nothing else; an S3-compatible UploadPart API (planned directly on top of this layer) receives parts out of order, numbered, each needing its own ETag. Building the sequential-only version means building the staging layer twice. The numbered store serves both: the sequential protocol is the special case where parts arrive in order, and the per-part md5 makes the S3 composite ETag (md5(concat(part md5s))-N) computable without re-reading the data.
The offset contract ¶
Offset is the total size of the CONTIGUOUS RUN OF PARTS FROM PART 1. It is the resume point and it is authoritative: a client that lost its state asks for the manifest and continues from there. Parts written out of order beyond a hole are kept but do not move the offset until the hole is filled.
Durability ¶
A part is written to a temporary file, fsynced, and only then renamed into place and recorded in the manifest — so a connection that dies mid-chunk leaves the offset exactly where it was, which is the whole point of resume. The manifest is rewritten the same way (temp + rename), so it is never observed half-written.
Index ¶
- Constants
- Variables
- func ValidID(id string) bool
- type Area
- func (a *Area) Create(id string, totalSize, chunkSize int64, hash string) (*Manifest, error)
- func (a *Area) CreateVariable(id string) (*Manifest, error)
- func (a *Area) Dir(id string) string
- func (a *Area) Enabled() bool
- func (a *Area) EnsureFree(size int64) error
- func (a *Area) Idle(id string, now time.Time) (time.Duration, error)
- func (a *Area) List() ([]string, error)
- func (a *Area) Manifest(id string) (*Manifest, error)
- func (a *Area) Open(id string) (*Reader, error)
- func (a *Area) Remove(id string) error
- func (a *Area) Root() string
- func (a *Area) Sweep(ttl time.Duration, now time.Time, keep func(id string) bool) ([]string, error)
- func (a *Area) Usage() (uploads int, bytes int64)
- func (a *Area) WritePart(id string, n int, r io.Reader, want int64) (*Manifest, error)
- func (a *Area) WriteVariablePart(id string, n int, r io.Reader, max int64) (*Manifest, error)
- type Manifest
- type Part
- type Reader
Constants ¶
const DiskHeadroomFactor = 1.2
DiskHeadroomFactor is the multiple of the declared size that must be free on the staging filesystem before an upload is accepted. 1.2 leaves room for the manifest, for filesystem overhead, and for the fact that other uploads are landing at the same time.
const ManifestName = "manifest.json"
ManifestName is the per-upload metadata file inside the staging directory.
const ManifestVersion = 1
ManifestVersion is bumped when the on-disk shape changes incompatibly.
const MaxParts = 10000
MaxParts mirrors S3's own multipart ceiling. Keeping the staging store inside the same limit means a staged upload can always be replayed as an S3 multipart upload without re-chunking for count reasons.
Variables ¶
var ( ErrNotFound = errors.New("staging: upload not found") ErrBadID = errors.New("staging: invalid upload id") ErrBadPart = errors.New("staging: invalid part") ErrShortPart = errors.New("staging: short part body") ErrIncomplete = errors.New("staging: upload incomplete") ErrNoDiskSpace = errors.New("staging: not enough free disk space") ErrTooManyParts = errors.New("staging: too many parts") )
Errors returned by the area. Callers map these onto status codes.
Functions ¶
Types ¶
type Area ¶
type Area struct {
// FreeBytes probes free space on the staging filesystem. Swapped in tests
// to exercise the disk guard without filling a disk.
FreeBytes func(dir string) (uint64, error)
// contains filtered or unexported fields
}
Area is a staging root plus the per-upload write locks.
Reads (manifest, assembled body) take no lock: every file is replaced by rename, so a reader either sees the old file or the new one, never a torn one.
func New ¶
New returns an Area rooted at dir. An empty dir yields a disabled area — Enabled() reports false and callers must refuse staged uploads rather than silently writing somewhere unexpected.
func (*Area) CreateVariable ¶
CreateVariable starts a staging directory whose parts may be any size — the shape S3 multipart needs. See Manifest.Variable for why both modes live in one area.
func (*Area) EnsureFree ¶
EnsureFree is the disk guard: it refuses when the staging filesystem has less than size*DiskHeadroomFactor available. A probe that cannot answer (unknown platform, permission) does NOT refuse — a guard that blocks all uploads because it cannot measure is worse than no guard.
func (*Area) Idle ¶
Idle reports how long ago the upload last saw activity, measured from the manifest's mtime (rewritten on every accepted part).
func (*Area) Open ¶
Open returns a reader over the assembled upload: every part concatenated in part-number order. The reader is seekable, so a driver (or the S3 SDK) can measure and replay the body, and so chunk 5 can serve a byte range out of staging while the transfer runs.
Refuses an incomplete upload — assembling over a hole would hand the driver a silently truncated file.
func (*Area) Sweep ¶
Sweep removes staging directories idle for longer than ttl. keep, when non-nil, is consulted first: returning true protects an id (the caller's DB says it is still live, e.g. a failed transfer being kept for retry).
Returns the ids it removed so the caller can log them — a sweep nobody can see is how 29 GB of temp files went unnoticed once already.
func (*Area) Usage ¶
Usage reports what is physically in the staging area right now: how many upload directories, and how many bytes of parts they hold.
It measures the DISK rather than trusting a counter, and that is the point. The in-flight/staged-bytes metrics are moved by events (begin, chunk, commit, abort) so they are fresh between sweeps, then reconciled against this on every sweeper pass — so a restart, or a crash that left parts behind, cannot leave the dashboard lying about how full the staging filesystem is.
func (*Area) WritePart ¶
WritePart stores part n from r.
want is the exact number of bytes the part must contain. A body shorter than want (the client's connection dropped mid-chunk) is DISCARDED and ErrShortPart returned: the offset must never advance over bytes we do not have. A body longer than want is also refused.
Re-writing an existing part is allowed and idempotent — a client that is not sure whether its last chunk landed simply sends it again.
func (*Area) WriteVariablePart ¶
WriteVariablePart stores part n with whatever length arrives, up to max bytes. Re-writing a part is idempotent — S3 clients retry a part by sending it again, and the last one written wins, exactly as S3 specifies.
⚠ Unlike WritePart there is no expected length to check against, so a truncated body cannot be detected HERE. That is not a gap: S3 puts the integrity check at CompleteMultipartUpload, where the client sends back the ETag of every part it believes it uploaded, and a part whose md5 disagrees is refused there.
type Manifest ¶
type Manifest struct {
Version int `json:"version"`
ID string `json:"id"`
TotalSize int64 `json:"total_size"`
ChunkSize int64 `json:"chunk_size"`
Hash string `json:"hash,omitempty"`
// Variable switches off the fixed grid.
//
// ⚠ The two modes exist because the two protocols disagree. filex's own
// resumable upload knows the total size up front and cuts it into equal
// chunks, so a part's exact length is checkable and a short chunk can be
// refused. S3 multipart knows neither: parts arrive with arbitrary sizes
// and the total is only known at CompleteMultipartUpload. Forcing S3 onto
// the fixed grid would mean rejecting perfectly legal uploads; giving S3
// its own staging area would mean two sweepers, two disk guards and two
// GCs. So one area, two modes.
Variable bool `json:"variable,omitempty"`
Parts []Part `json:"parts"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
Manifest is the on-disk record of what is physically staged. It is the authority for the resume offset; the DB row mirrors it so listings and the sweeper do not have to open every manifest.
func (*Manifest) Complete ¶
Complete reports whether every part is present and the sizes add up to the declared total.
func (*Manifest) CompositeETag ¶
CompositeETag is the S3-style multipart ETag for the staged parts: md5(concat(raw part md5s))-N. Computed from the manifest alone — the future S3 gateway can answer with it without re-reading a byte.
func (*Manifest) ExpectedPartSize ¶
ExpectedPartSize is the exact length part n must have: a full chunk, or the remainder for the final part. Returns 0 for an out-of-range number.
func (*Manifest) Offset ¶
Offset is the resume point: the total size of the contiguous run of parts starting at part 1. A part written past a hole does not count until the hole is filled — which is exactly what a sequential client needs to be told.
type Part ¶
Part is one staged chunk: its number (1-based), its byte length and the md5 of its bytes.
type Reader ¶
type Reader struct {
// contains filtered or unexported fields
}
Reader concatenates the part files of a staged upload into one seekable stream. Only one part file is open at a time.
Source Files