upload

type BucketSpec ¶

type BucketSpec struct {
	Index int   `json:"index"`
	Dims  []int `json:"dims"`
}

BucketSpec is one bounded .pt2 input-shape declaration. Every bucket in a manifest carries the same complete set of input indexes.

type Digest ¶

type Digest struct {
	Size   int64
	CRC32C string // base64 of the big-endian 4-byte sum (GCS convention)
	SHA256 string // lowercase hex
}

Digest holds the size and content digests of a local file.

func DigestFile(path string) (Digest, error)

type FileSpec ¶

type FileSpec struct {
	ClientFileID string
	Path         string // local filesystem path
	Role         string
	InputIndex   int  // -1 unless Role == RoleInput
	BucketIndex  *int // nil unless this is a bucketed input
	Filename     string
	Size         int64
	CRC32C       string
	SHA256       string
}

FileSpec is one fully-digested manifest entry plus the local path it came from. Specs are ordered: model first, then inputs (InputIndex order), then external data; ClientFileID is "f<position>" in that order.

func BuildBucketedManifest(model string, inputs, external []string, buckets []BucketSpec, note NoteFunc) ([]FileSpec, error)

func BuildManifest(model string, inputs, external []string, note NoteFunc) ([]FileSpec, error)

func LoadManifestDoc(path string, note NoteFunc) ([]FileSpec, error)

type NoteFunc ¶

type NoteFunc func(path string, size int64)

NoteFunc observes a file about to be digested (used for "hashing large file" progress messages). size is from os.Stat, before the read pass.

type SessionLease ¶

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

SessionLease holds the cross-process lock for one upload session. Commands retain it across state reconciliation, transfer, completion, or cancellation.

func AcquireSession(ctx context.Context, sessionID string) (*SessionLease, error)

func (l *SessionLease) Close() error

type State ¶

type State struct {
	SessionID string       `json:"session_id"`
	Repo      string       `json:"repo"` // ACCOUNT/NAME
	Tag       string       `json:"tag"`
	CreatedAt time.Time    `json:"created_at"`
	Files     []*StateFile `json:"files"`
}

State is the persisted record of an in-flight upload session, stored as <StateDir>/<session-id>.json.

It contains resumable session URIs, which are bearer credentials: the file is written 0600 in a 0700 directory and its contents must never be printed or logged.

func LoadState(sessionID string) (*State, error)

func (s *State) Save() error

type StateFile ¶

type StateFile struct {
	ClientFileID  string `json:"client_file_id"`
	LocalPath     string `json:"local_path"`
	CanonicalPath string `json:"canonical_path"`
	UploadURL     string `json:"upload_url,omitempty"`  // signed resumable-start URL
	SessionURI    string `json:"session_uri,omitempty"` // resumable session URI (credential)
	Size          int64  `json:"size"`
	CRC32C        string `json:"crc32c"`
	Offset        int64  `json:"offset"`
	Uploaded      bool   `json:"uploaded"`
}

StateFile tracks one manifest file's upload progress. Offset is a hint only — the GCS committed offset is authoritative on resume.

type Uploader ¶

type Uploader struct {
	Client       *http.Client
	ChunkSize    int64         // rounded up to a 256 KiB multiple; default 8 MiB
	StallTimeout time.Duration // per-chunk inactivity budget; 0 = no limit
	MaxRetries   int           // consecutive retryable failures tolerated per call (default 4)
	// Sleep blocks between retries; nil = timer sleep. Injectable for tests.
	Sleep func(ctx context.Context, d time.Duration) error
}

Uploader speaks the GCS resumable upload protocol against v4 signed resumable-start URLs.

It deliberately runs over a BARE http.Client: no Authorization header, no melange transport chain (auth/retry/debug) — GCS must never see the PAT, signed URLs carry their own credentials, and debug logging must never see signed URLs or session URIs. Upload-specific retry lives here instead: on retryable failures the committed offset is re-queried from the server and the transfer continues from there, never resending acknowledged bytes.

func (u *Uploader) QueryOffset(ctx context.Context, sessionURI string, total int64) (offset int64, done bool, err error)

func (u *Uploader) StartSession(ctx context.Context, uploadURL string) (string, error)

func (u *Uploader) UploadFile(ctx context.Context, sessionURI, path string, total, from int64, onCommit func(committed int64)) error