orders

Documentation

Index

• Variables
• type ArchiveRequest
• type ArchivedInfo
• type Config
• type DB
• type FileStore
  • func NewFileStore(ordersDir string, orderLimitGracePeriod, maxInFlightTime time.Duration) (*FileStore, error)
  • func (store *FileStore) Archive(satelliteID storj.NodeID, createdAtHour, archivedAt time.Time, ...) error
  • func (store *FileStore) CleanArchive(deleteBefore time.Time) error
  • func (store *FileStore) Enqueue(info *Info) (err error)
  • func (store *FileStore) ListArchived() ([]*ArchivedInfo, error)
  • func (store *FileStore) ListUnsentBySatellite() (infoMap map[storj.NodeID]UnsentInfo, err error)
  • func (store *FileStore) TestSetSettleBuffer(orderLimitGracePeriod, maxInFlightTime time.Duration)
• type Info
• type Service
  • func NewService(log *zap.Logger, dialer rpc.Dialer, orders DB, trust *trust.Pool, ...) *Service
  • func (service *Service) Close() error
  • func (service *Service) Run(ctx context.Context) (err error)
  • func (service *Service) Settle(ctx context.Context, satelliteID storj.NodeID, orders []*Info, ...)
• type Status
• type UnsentInfo

Variables

var (
	// OrderError represents errors with orders
	OrderError = errs.Class("order")
	// OrderNotFoundError is the error returned when an order is not found
	OrderNotFoundError = errs.Class("order not found")
)

Types

type ArchiveRequest

type ArchiveRequest struct {
	Satellite storj.NodeID
	Serial    storj.SerialNumber
	Status    Status
}

ArchiveRequest defines arguments for archiving a single order.

type ArchivedInfo

type ArchivedInfo struct {
	Limit *pb.OrderLimit
	Order *pb.Order

	Status     Status
	ArchivedAt time.Time
}

ArchivedInfo contains full information about an archived order.

type Config

type Config struct {
	MaxSleep          time.Duration `help:"maximum duration to wait before trying to send orders" releaseDefault:"300s" devDefault:"1s"`
	SenderInterval    time.Duration `help:"duration between sending" releaseDefault:"1h0m0s" devDefault:"30s"`
	SenderTimeout     time.Duration `help:"timeout for sending" default:"1h0m0s"`
	SenderDialTimeout time.Duration `help:"timeout for dialing satellite during sending orders" default:"1m0s"`
	CleanupInterval   time.Duration `help:"duration between archive cleanups" default:"1h0m0s"`
	ArchiveTTL        time.Duration `help:"length of time to archive orders before deletion" default:"168h0m0s"` // 7 days
}

Config defines configuration for sending orders.

type DB

type DB interface {
	// Enqueue inserts order to the list of orders needing to be sent to the satellite.
	Enqueue(ctx context.Context, info *Info) error
	// ListUnsent returns orders that haven't been sent yet.
	ListUnsent(ctx context.Context, limit int) ([]*Info, error)
	// ListUnsentBySatellite returns orders that haven't been sent yet grouped by satellite.
	ListUnsentBySatellite(ctx context.Context) (map[storj.NodeID][]*Info, error)

	// Archive marks order as being handled.
	Archive(ctx context.Context, archivedAt time.Time, requests ...ArchiveRequest) error
	// ListArchived returns orders that have been sent.
	ListArchived(ctx context.Context, limit int) ([]*ArchivedInfo, error)
	// CleanArchive deletes all entries older than ttl
	CleanArchive(ctx context.Context, ttl time.Duration) (int, error)
}

DB implements storing orders for sending to the satellite.

architecture: Database

type FileStore

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

FileStore implements the orders.Store interface by appending orders to flat files.

func NewFileStore

func NewFileStore(ordersDir string, orderLimitGracePeriod, maxInFlightTime time.Duration) (*FileStore, error)

NewFileStore creates a new orders file store, and the directories necessary for its use.

func (*FileStore) Archive

func (store *FileStore) Archive(satelliteID storj.NodeID, createdAtHour, archivedAt time.Time, status pb.SettlementWithWindowResponse_Status) error

Archive moves a file from "unsent" to "archive". The filename/path changes from unsent/unsent-orders-<satelliteID>-<createdAtHour> to archive/archived-orders-<satelliteID>-<createdAtHour>-<archivedTime>-<ACCEPTED/REJECTED>.

func (*FileStore) CleanArchive

func (store *FileStore) CleanArchive(deleteBefore time.Time) error

CleanArchive deletes all entries archvied before the provided time.

func (*FileStore) Enqueue

func (store *FileStore) Enqueue(info *Info) (err error)

Enqueue inserts order to be sent at the end of the unsent file for a particular creation hour. It assumes the order is not being queued after the order limit grace period.

func (*FileStore) ListArchived

func (store *FileStore) ListArchived() ([]*ArchivedInfo, error)

ListArchived returns orders that have been sent.

func (*FileStore) ListUnsentBySatellite

func (store *FileStore) ListUnsentBySatellite() (infoMap map[storj.NodeID]UnsentInfo, err error)

ListUnsentBySatellite returns one window of orders that haven't been sent yet, grouped by satellite. It only reads files where the order limit grace period has passed, meaning no new orders will be appended. There is a separate window for each created at hour, so if a satellite has 2 windows, `ListUnsentBySatellite` needs to be called twice, with calls to `Archive` in between each call, to see all unsent orders.

func (*FileStore) TestSetSettleBuffer

func (store *FileStore) TestSetSettleBuffer(orderLimitGracePeriod, maxInFlightTime time.Duration)

TestSetSettleBuffer is a function that allows us to modify order limit grace period and max inflight time for testing purposes.

type Info

type Info struct {
	Limit *pb.OrderLimit
	Order *pb.Order
}

Info contains full information about an order.

type Service

type Service struct {
	Sender  *sync2.Cycle
	Cleanup *sync2.Cycle
	// contains filtered or unexported fields
}

Service sends every interval unsent orders to the satellite.

architecture: Chore

func NewService

func NewService(log *zap.Logger, dialer rpc.Dialer, orders DB, trust *trust.Pool, config Config) *Service

NewService creates an order service.

func (*Service) Close

func (service *Service) Close() error

Close stops the sending service.

func (*Service) Run

func (service *Service) Run(ctx context.Context) (err error)

Run sends orders on every interval to the appropriate satellites.

func (*Service) Settle

func (service *Service) Settle(ctx context.Context, satelliteID storj.NodeID, orders []*Info, requests chan ArchiveRequest)

Settle uploads orders to the satellite.

type Status

type Status byte

Status is the archival status of the order.

const (
	StatusUnsent Status = iota
	StatusAccepted
	StatusRejected
)

Statuses for satellite responses.

type UnsentInfo

type UnsentInfo struct {
	CreatedAtHour time.Time
	InfoList      []*Info
}

UnsentInfo is a struct containing a window of orders for a satellite and order creation hour.