orders

Documentation

Overview

Package orders handles order limit management. It contains file database for the orders. Similarly, it manages settlement, where order limits are sent to the satellite.

Index

• Variables
• type ArchiveRequest
• type ArchivedInfo
• type Config
• type DB
• type FileStore
  • func NewFileStore(log *zap.Logger, ordersDir string, orderLimitGracePeriod time.Duration) (*FileStore, error)
  • func (store *FileStore) Archive(satelliteID storj.NodeID, unsentInfo UnsentInfo, archivedAt time.Time, ...) error
  • func (store *FileStore) BeginEnqueue(satelliteID storj.NodeID, createdAt time.Time) (commit func(*ordersfile.Info) error, err error)
  • func (store *FileStore) CleanArchive(deleteBefore time.Time) error
  • func (store *FileStore) Enqueue(info *ordersfile.Info) (err error)
  • func (store *FileStore) ListArchived() ([]*ArchivedInfo, error)
  • func (store *FileStore) ListUnsentBySatellite(ctx context.Context, now time.Time) (infoMap map[storj.NodeID]UnsentInfo, err error)
• type Service
  • func NewService(log *zap.Logger, dialer rpc.Dialer, ordersStore *FileStore, orders DB, ...) *Service
  • func (service *Service) CleanArchive(ctx context.Context, deleteBefore time.Time) (err error)
  • func (service *Service) Close() error
  • func (service *Service) Run(ctx context.Context) (err error)
  • func (service *Service) SendOrders(ctx context.Context, now time.Time)
  • func (service *Service) TestSetLogger(log *zap.Logger)
• 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:"30s" 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:"5m0s"`
	ArchiveTTL        time.Duration `help:"length of time to archive orders before deletion" default:"168h0m0s"` // 7 days
	Path              string        `help:"path to store order limit files in" default:"$CONFDIR/orders"`
}

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 *ordersfile.Info) error
	// ListUnsent returns orders that haven't been sent yet.
	ListUnsent(ctx context.Context, limit int) ([]*ordersfile.Info, error)
	// ListUnsentBySatellite returns orders that haven't been sent yet grouped by satellite.
	ListUnsentBySatellite(ctx context.Context) (map[storj.NodeID][]*ordersfile.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 the before time.
	CleanArchive(ctx context.Context, deleteBefore time.Time) (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(log *zap.Logger, ordersDir string, orderLimitGracePeriod 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, unsentInfo UnsentInfo, archivedAt time.Time, status pb.SettlementWithWindowResponse_Status) error

Archive moves a file from "unsent" to "archive".

func (*FileStore) BeginEnqueue

func (store *FileStore) BeginEnqueue(satelliteID storj.NodeID, createdAt time.Time) (commit func(*ordersfile.Info) error, err error)

BeginEnqueue returns a function that can be called to enqueue the passed in Info. If the Info is too old to be enqueued, then an error is returned.

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 *ordersfile.Info) (err error)

Enqueue inserts order to be sent at the end of the unsent file for a particular creation hour. It ensures 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(ctx context.Context, now time.Time) (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.

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, ordersStore *FileStore, orders DB, trust *trust.Pool, config Config) *Service

NewService creates an order service.

func (*Service) CleanArchive

func (service *Service) CleanArchive(ctx context.Context, deleteBefore time.Time) (err error)

CleanArchive removes all archived orders that were archived before the deleteBefore time.

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) SendOrders

func (service *Service) SendOrders(ctx context.Context, now time.Time)

SendOrders sends the orders using now as the current time.

func (*Service) TestSetLogger

func (service *Service) TestSetLogger(log *zap.Logger)

TestSetLogger sets the logger.

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
	Version       ordersfile.Version
	InfoList      []*ordersfile.Info
}

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