transfers

README

DTS Transfer Orchestration

The transfer package implements the orchestration logic that DTS uses to perform file transfers. It consists of a family of concurrent sequential programs, each implemented by a singleton struct backed by a goroutine:

• The dispatcher accepts transfer-related requests from "clients" (properly-routed requests from the web service layer) and dispatching them to other programs.
• The store maintains the state of transfer records, allowing other programs to fetch transfer metadata and manipulate status information.
• The stager handles file staging operations, initiating file transfers via the mover on success.
• The mover handles the transfer of the file payloads, initiating the generation and transfer of manifests via the manifestor on success.
• The manifestor generates metadata (JSON) manifests for completed transfers, moving each into place with the rest of its payload.

Here's a sequence diagram illustrating what happens when everything goes right:

More detailed documentation is available in the source files for these programs.

Documentation

Index

• Constants
• func Cancel(transferId uuid.UUID, orcid string) error
• func Create(spec Specification) (uuid.UUID, error)
• func Running() bool
• func Start(conf config.Config) error
• func Stop() error
• func Unsubscribe(sub Subscription) error
• type AlreadyRunningError
  • func (t AlreadyRunningError) Error() string
• type Database
• type Endpoint
• type FileTransfer
• type InvalidDatabaseConfigError
  • func (e InvalidDatabaseConfigError) Error() string
• type InvalidOrcidError
  • func (e InvalidOrcidError) Error() string
• type Message
• type NoDatabasesAvailable
  • func (t NoDatabasesAvailable) Error() string
• type NoFilesRequestedError
  • func (t NoFilesRequestedError) Error() string
• type NotRunningError
  • func (t NotRunningError) Error() string
• type PayloadTooLargeError
  • func (e PayloadTooLargeError) Error() string
• type SaveFileError
  • func (e SaveFileError) Error() string
• type Specification
• type Subscription
  • func Subscribe(bufferSize int) Subscription
• type TransferNotFoundError
  • func (e TransferNotFoundError) Error() string
• type TransferStatus
  • func Status(transferId uuid.UUID) (TransferStatus, error)
• type TransferStatusCode

Constants

const (
	TransferStatusUnknown    = endpoints.TransferStatusUnknown
	TransferStatusStaging    = endpoints.TransferStatusStaging
	TransferStatusActive     = endpoints.TransferStatusActive
	TransferStatusFailed     = endpoints.TransferStatusFailed
	TransferStatusFinalizing = endpoints.TransferStatusFinalizing
	TransferStatusInactive   = endpoints.TransferStatusInactive
	TransferStatusSucceeded  = endpoints.TransferStatusSucceeded
)

useful constants

Functions

func Cancel

func Cancel(transferId uuid.UUID, orcid string) error

Requests that the transfer with the given UUID (initiated by the user with the given Orcid) be canceled. Clients should check the status of the transfer separately.

func Create

func Create(spec Specification) (uuid.UUID, error)

Creates a new transfer associated with the user with the specified Orcid ID to the manager's set, returning a UUID for the transfer. The transfer is defined by specifying the names of the source and destination databases and a set of file IDs associated with the source.

func Running

func Running() bool

Returns true if transfers are currently being processed, false if not.

func Start

func Start(conf config.Config) error

starts processing transfers according to the given configuration, returning an informative error if anything prevents this

func Stop

func Stop() error

Stops processing transfers. Adding new transfers and requesting transfer statuses are disallowed in a stopped state.

func Unsubscribe

func Unsubscribe(sub Subscription) error

Unsubscribes the caller from its messages.

Types

type AlreadyRunningError

type AlreadyRunningError struct{}

indicates that Start() has been called when transfers are being processed

func (AlreadyRunningError) Error

func (t AlreadyRunningError) Error() string

type Database

type Database = databases.Database

useful type aliases

type Endpoint

type Endpoint = endpoints.Endpoint

type FileTransfer

type FileTransfer = endpoints.FileTransfer

type InvalidDatabaseConfigError

type InvalidDatabaseConfigError struct {
	Database, Message string
}

Indicates an invalid configuration for databases

func (InvalidDatabaseConfigError) Error

func (e InvalidDatabaseConfigError) Error() string

type InvalidOrcidError

type InvalidOrcidError struct {
	Orcid   string
	Message string
}

func (InvalidOrcidError) Error

func (e InvalidOrcidError) Error() string

type Message

type Message struct {
	Description    string
	TransferId     uuid.UUID
	TransferStatus TransferStatus
	Time           time.Time
}

A message sent to a subscriber relating to a transfer

type NoDatabasesAvailable

type NoDatabasesAvailable struct{}

indicates that no databases are currently available

func (NoDatabasesAvailable) Error

func (t NoDatabasesAvailable) Error() string

type NoFilesRequestedError

type NoFilesRequestedError struct{}

indicates that a transfer has been requested with no files(!)

func (NoFilesRequestedError) Error

func (t NoFilesRequestedError) Error() string

type NotRunningError

type NotRunningError struct{}

indicates that Stop() has been called when transfers are not being processed

func (NotRunningError) Error

func (t NotRunningError) Error() string

type PayloadTooLargeError

type PayloadTooLargeError struct {
	Size float64 // size of the requested payload in gigabytes
}

indicates that a payload has been requested that is too large

func (PayloadTooLargeError) Error

func (e PayloadTooLargeError) Error() string

type SaveFileError

type SaveFileError struct {
	Filename string
	Message  string
}

indicates an error encountered in saving transfer data to a persistent file for loading later

func (SaveFileError) Error

func (e SaveFileError) Error() string

type Specification

type Specification struct {
	// a Markdown description of the transfer transfer
	Description string
	// the name of destination database to which files are transferred (as specified in the config
	// file) OR a custom destination spec (<provider>:<id>:<credential>)
	Destination string
	// machine-readable instructions for processing the payload at its destination
	Instructions map[string]any
	// an array of identifiers for files to be transferred from Source to Destination
	FileIds []string
	// the name of source database from which files are transferred (as specified in the config file)
	Source string
	// the time at which the transfer is requested
	TimeOfRequest time.Time
	// information about the user requesting the transfer
	User auth.User
}

this type holds a specification used to create a valid transfer transfer

type Subscription

type Subscription struct {
	Id      int
	Channel <-chan Message
}

A subscription that identifies the subscriber and provides it with a channel on which to receive messages

func Subscribe

func Subscribe(bufferSize int) Subscription

Subscribes the caller to the transfer message feed, returning a channel with the given buffer size on which messages are queued.

type TransferNotFoundError

type TransferNotFoundError struct {
	Id uuid.UUID
}

indicates that the transfer with a given ID is not found

func (TransferNotFoundError) Error

func (e TransferNotFoundError) Error() string

type TransferStatus

type TransferStatus = endpoints.TransferStatus

func Status

func Status(transferId uuid.UUID) (TransferStatus, error)

Given a transfer ID, returns its transfer status (or a non-nil error indicating any issues encountered).

type TransferStatusCode

type TransferStatusCode = endpoints.TransferStatusCode