upload

package
v2.0.0-...-c48d65c Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 10, 2026 License: Apache-2.0 Imports: 36 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var RegisteredEvents = []events.Unmarshaller{
	events.PostprocessingFinished{},
	events.PostprocessingStepFinished{},
	events.RestartPostprocessing{},
	events.CleanUpload{},
}

RegisteredEvents are the postprocessing events the coordinator consumes.

Functions

func NewCoordinator 

func NewCoordinator(fs storage.FS, store SessionStore, chunkFolder string, pub events.Publisher) *coordinator

NewCoordinator constructs a coordinator backed by the given driver and store.

Types

type AsyncConf 

type AsyncConf struct {
	Enabled       bool
	ConsumerGroup string
	NumConsumers  int
	// MountID is the storage id this provider answers for, used to drop
	// postprocessing events belonging to other storages.
	MountID string
}

AsyncConf is how a service asks for async uploads: whether they are enabled, and the consumer subscription to use if they are.

func AsyncConfFromDriverConf 

func AsyncConfFromDriverConf(driverConf map[string]interface{}) AsyncConf

AsyncConfFromDriverConf reads the postprocessing settings off the driver's own config keys, so the coordinator and the driver cannot disagree about them.

type Coordinator 

type Coordinator interface {
	// InitiateUpload returns the protocols and ids that bytes can be appended to.
	InitiateUpload(ctx context.Context, ref *provider.Reference, uploadLength int64, metadata map[string]string) (map[string]string, error)
	// GetUpload returns the session with the given id as a tusd upload.
	GetUpload(ctx context.Context, id string) (tusd.Upload, error)
	// UseIn registers the coordinator as the tusd data store.
	UseIn(composer *tusd.StoreComposer)
	// ListUploadSessions returns the upload sessions matching the given filter.
	ListUploadSessions(ctx context.Context, filter storage.UploadSessionFilter) ([]storage.UploadSession, error)
	// Upload writes the whole body of a non-resumable (PUT) upload and finishes it.
	Upload(ctx context.Context, req storage.UploadRequest, uff storage.UploadFinishedFunc) (*provider.ResourceInfo, error)
	// StartPostprocessing subscribes to postprocessing results and enables async
	// uploads. Call once, before serving requests.
	StartPostprocessing(stream events.Consumer, group, mountID string, numConsumers int) error
}

Coordinator owns the upload lifecycle: initiation, data transfer and listing.

type FileSession 

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

FileSession is the Session implementation for disk-backed uploads. While an upload is in progress, incoming bytes are staged in a .bin file and upload metadata (size, owner, checksums, etc.) is persisted in a .info file. Both survive process restarts, allowing TUS resumption.

In scope: read/write the staged .bin file, persist/load upload metadata in the .info file. Out of scope: TUS protocol, checksums, event publishing, postprocessing — those live in coordinatedUpload and coordinator.

func (*FileSession) BinPath 

func (s *FileSession) BinPath() string

BinPath returns the path to the staged binary file.

func (*FileSession) Checksums 

func (s *FileSession) Checksums() storage.UploadChecksums

Checksums returns the pre-computed checksums stored on the session.

func (*FileSession) Chunk 

func (s *FileSession) Chunk() string

Chunk returns the chunk basename stored in the session, or "" for non-chunked uploads.

func (*FileSession) Cleanup 

func (s *FileSession) Cleanup(ctx context.Context, cleanBin, cleanInfo bool)

Cleanup removes the staged binary and/or info file. Node deletion and processing flag changes are the coordinator's responsibility.

func (*FileSession) Context 

func (s *FileSession) Context(ctx context.Context) context.Context

Context reconstructs a context carrying the user, logger, lock ID, and initiator ID that were recorded when the upload was initiated.

func (*FileSession) Dir 

func (s *FileSession) Dir() string

Dir returns the directory portion of the upload path.

func (*FileSession) Executant 

func (s *FileSession) Executant() userpb.UserId

Executant returns the user ID of the user who initiated this upload.

func (*FileSession) ExecutantUser 

func (s *FileSession) ExecutantUser() *userpb.User

ExecutantUser returns the full identity of the user who initiated the upload. Upload events must carry it rather than the bare id from Executant(): consumers read the display name straight off the event and do not look it up, so an id-only user reaches the activity feed with a blank name.

func (*FileSession) Expires 

func (s *FileSession) Expires() time.Time

Expires returns the upload expiry time.

func (*FileSession) Filename 

func (s *FileSession) Filename() string

Filename returns the filename stored in the session.

func (*FileSession) GetInfo 

func (s *FileSession) GetInfo(_ context.Context) (tusd.FileInfo, error)

func (*FileSession) GetReader 

func (s *FileSession) GetReader(_ context.Context) (io.ReadCloser, error)

func (*FileSession) ID 

func (s *FileSession) ID() string

ID returns the upload session ID.

func (*FileSession) InitiatorID 

func (s *FileSession) InitiatorID() string

func (*FileSession) IsProcessing 

func (s *FileSession) IsProcessing() bool

IsProcessing returns true if all bytes are received but postprocessing has not finished.

func (*FileSession) Metadata 

func (s *FileSession) Metadata() map[string]string

Metadata returns the upload metadata the coordinator passes to the driver.

func (*FileSession) NodeExists 

func (s *FileSession) NodeExists() bool

NodeExists returns whether the target node existed when the upload was initiated.

func (*FileSession) NodeID 

func (s *FileSession) NodeID() string

NodeID returns the node ID for this upload.

func (*FileSession) NodeParentID 

func (s *FileSession) NodeParentID() string

NodeParentID returns the parent node ID for this upload.

func (*FileSession) Offset 

func (s *FileSession) Offset() int64

Offset returns the current upload offset.

func (*FileSession) Persist 

func (s *FileSession) Persist(ctx context.Context) error

Persist writes the session metadata atomically to disk.

func (*FileSession) ProviderID 

func (s *FileSession) ProviderID() string

ProviderID returns the storage provider ID stored in the session.

func (*FileSession) Purge 

func (s *FileSession) Purge(ctx context.Context)

Purge removes all on-disk state for this session.

func (*FileSession) Reference 

func (s *FileSession) Reference() provider.Reference

Reference returns a CS3 reference for the resource being uploaded.

func (*FileSession) ScanData 

func (s *FileSession) ScanData() (string, time.Time)

ScanData returns the AV scan result and scan date stored on the session.

func (*FileSession) SetChecksums 

func (s *FileSession) SetChecksums(sha1Sum, md5Sum, adler32Sum []byte)

SetChecksums stores pre-computed checksums so CommitUpload can use them without re-reading the binary file.

func (*FileSession) SetExecutant 

func (s *FileSession) SetExecutant(u *userpb.User)

SetExecutant stores the identity of the user who initiated the upload.

func (*FileSession) SetMetadata 

func (s *FileSession) SetMetadata(key, value string)

SetMetadata sets a user-visible upload metadata field.

func (*FileSession) SetScanData 

func (s *FileSession) SetScanData(result string, date time.Time)

SetScanData stores AV scan results on the session.

func (*FileSession) SetSize 

func (s *FileSession) SetSize(size int64)

SetSize updates the declared upload size.

func (*FileSession) SetSizeDiff 

func (s *FileSession) SetSizeDiff(d int64)

SetSizeDiff records the size change PrepareUpload reported. It is persisted because the async path prepares and commits in different processes, so the value cannot be held in memory between the two.

func (*FileSession) SetSizeIsDeferred 

func (s *FileSession) SetSizeIsDeferred(value bool)

SetSizeIsDeferred marks the upload size as not yet known.

func (*FileSession) SetStorageValue 

func (s *FileSession) SetStorageValue(key, value string)

SetStorageValue sets an internal storage field on the session.

func (*FileSession) SetVersionCreated 

func (s *FileSession) SetVersionCreated(v bool)

SetVersionCreated records what PrepareUpload reported. Persisted for the same reason as the size diff: on the async path the commit runs in another process.

func (*FileSession) Size 

func (s *FileSession) Size() int64

Size returns the declared upload size.

func (*FileSession) SizeDiff 

func (s *FileSession) SizeDiff() int64

SizeDiff returns the tree size change PrepareUpload propagated optimistically. Rolling an upload back has to undo exactly that amount.

func (*FileSession) SpaceGid 

func (s *FileSession) SpaceGid() string

SpaceGid returns the numeric GID of the space owner, or "" if not set.

func (*FileSession) SpaceID 

func (s *FileSession) SpaceID() string

SpaceID returns the space (root) ID.

func (*FileSession) SpaceOwner 

func (s *FileSession) SpaceOwner() *userpb.UserId

SpaceOwner returns the space owner user ID.

func (*FileSession) ToFileInfo 

func (s *FileSession) ToFileInfo() tusd.FileInfo

func (*FileSession) TouchBin 

func (s *FileSession) TouchBin() error

TouchBin creates the empty staging file.

func (*FileSession) URL 

func (s *FileSession) URL(_ context.Context) (string, error)

URL returns a signed JWT URL that the postprocessing service can use to download the staged binary via the data gateway.

func (*FileSession) VersionCreated 

func (s *FileSession) VersionCreated() bool

VersionCreated reports whether this upload superseded existing content, which UploadReady consumers use to tell an overwrite from a new file.

func (*FileSession) WriteChunk 

func (s *FileSession) WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error)

type FileStore 

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

FileStore is a filesystem-backed SessionStore. Sessions are stored as a pair of files in the upload directory:

  • <id>.info — JSON-encoded tusd.FileInfo
  • <id> — staged binary bytes

This is the same on-disk format used by OcisStore so existing sessions survive a rolling deploy that switches to FileStore.

func FileStoreFromDriverConf 

func FileStoreFromDriverConf(driverConf map[string]interface{}, log *zerolog.Logger) *FileStore

FileStoreFromDriverConf builds a FileStore from a reva driver config map. Returns nil if the config carries no upload path (driver does not support coordinated uploads). Each service that mounts the same driver calls this independently.

func NewFileStore 

func NewFileStore(uploadDir string, opts TokenOptions, log *zerolog.Logger) *FileStore

NewFileStore creates a FileStore staging uploads in uploadDir. uploadDir must be on a shared filesystem when multiple pods handle the same space.

func NewFileStoreFromConfig 

func NewFileStoreFromConfig(uploadDir string, driverConf map[string]interface{}, log *zerolog.Logger) *FileStore

NewFileStoreFromConfig builds a FileStore staging uploads in uploadDir when set, falling back to the active driver config. This allows drivers that have no local root (e.g. KW) to still get a coordinator by setting upload_directory at the service level rather than inside the driver. Returns nil only when neither source resolves to a non-empty path.

func (*FileStore) Get 

func (fs *FileStore) Get(ctx context.Context, id string) (Session, error)

Get loads the session with the given id from disk.

func (*FileStore) List 

func (fs *FileStore) List(ctx context.Context) ([]Session, error)

List returns all sessions found in the upload directory.

func (*FileStore) New 

func (fs *FileStore) New(_ context.Context) Session

New allocates a fresh session with a new UUID.

func (*FileStore) Setup 

func (fs *FileStore) Setup() error

Setup creates the upload directory eagerly so permission problems are caught at startup rather than on the first upload.

func (*FileStore) UploadDir 

func (fs *FileStore) UploadDir() string

UploadDir returns the directory this FileStore stages uploads in.

type Session 

type Session interface {
	storage.UploadSession

	// Data access — delegated to by coordinatedUpload for TUS reads/writes.
	GetInfo(ctx context.Context) (tusd.FileInfo, error)
	GetReader(ctx context.Context) (io.ReadCloser, error)
	WriteChunk(ctx context.Context, offset int64, src io.Reader) (int64, error)

	// Internal coordinator plumbing.
	Chunk() string
	BinPath() string
	ExecutantUser() *userpb.User
	ProviderID() string
	SpaceID() string
	NodeID() string
	NodeParentID() string
	NodeExists() bool
	Dir() string
	URL(ctx context.Context) (string, error)
	SetScanData(result string, date time.Time)
	Checksums() storage.UploadChecksums
	SetChecksums(sha1, md5, adler32 []byte)
	SizeDiff() int64
	SetSizeDiff(d int64)
	VersionCreated() bool
	SetVersionCreated(v bool)
	Metadata() map[string]string
	Persist(ctx context.Context) error
	Cleanup(ctx context.Context, cleanBin, cleanInfo bool)
	Context(ctx context.Context) context.Context

	// Typed setters used by Coordinator.InitiateUpload to populate a new session
	// without knowing internal storage key names.
	SetStorageValue(key, value string)
	SetMetadata(key, value string)
	SetSize(size int64)
	SetSizeIsDeferred(value bool)
	SetExecutant(u *userpb.User)
	TouchBin() error
}

Session is the driver-agnostic view of an upload session the Coordinator needs. Implementations must be pure state (CRUD): protocol orchestration belongs to coordinatedUpload or the coordinator itself.

type SessionStore 

type SessionStore interface {
	New(ctx context.Context) Session
	Get(ctx context.Context, id string) (Session, error)
	List(ctx context.Context) ([]Session, error)
}

SessionStore abstracts upload-session persistence for the Coordinator.

type TokenOptions 

type TokenOptions struct {
	DownloadEndpoint     string
	DataGatewayEndpoint  string
	TransferSharedSecret string
	TransferExpires      int64
}

TokenOptions carries the JWT-signing configuration needed to produce transfer URLs for the postprocessing service.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.