sjstore

Documentation

Index

• Variables
• type ListConfig
• type Select
• type StartedDeals
• type Stats
• type Store
  • func New(ds datastore.TxnDatastore) (*Store, error)
  • func (s *Store) AddStartedDeals(iid ffs.APIID, c cid.Cid, proposals []cid.Cid) error
  • func (s *Store) CancelQueued(jid ffs.JobID) (bool, error)
  • func (s *Store) Close() error
  • func (s *Store) Dequeue(iid ffs.APIID) (*ffs.StorageJob, error)
  • func (s *Store) Enqueue(j ffs.StorageJob) error
  • func (s *Store) Finalize(jid ffs.JobID, st ffs.JobStatus, jobError error, dealErrors []ffs.DealError) error
  • func (s *Store) Get(jid ffs.JobID) (ffs.StorageJob, error)
  • func (s *Store) GetExecutingJob(iid ffs.APIID, c cid.Cid) *ffs.JobID
  • func (s *Store) GetExecutingJobIDs() []ffs.JobID
  • func (s *Store) GetStartedDeals(iid ffs.APIID, c cid.Cid) ([]cid.Cid, error)
  • func (s *Store) GetStats() Stats
  • func (s *Store) List(config ListConfig) ([]ffs.StorageJob, bool, string, error)
  • func (s *Store) MonitorJob(j ffs.StorageJob) chan deals.StorageDealInfo
  • func (s *Store) RemoveStartedDeals(iid ffs.APIID, c cid.Cid) error
  • func (s *Store) Watch(ctx context.Context, c chan<- ffs.StorageJob, iid ffs.APIID) error

Variables

var (

	// ErrNotFound indicates the instance doesn't exist.
	ErrNotFound = errors.New("job not found")
)

Types

type ListConfig

type ListConfig struct {
	// APIIDFilter filters StorageJobs list to the specified APIID. Defaults to no filter.
	APIIDFilter ffs.APIID
	// CidFilter filters StorageJobs list to the specified cid. Defaults to no filter.
	CidFilter cid.Cid
	// Limit limits the number of StorageJobs returned. Defaults to no limit.
	Limit uint64
	// Ascending returns the StorageJobs ascending by time. Defaults to false, descending.
	Ascending bool
	// Select specifies to return StorageJobs in the specified state.
	Select Select
	// NextPageToken sets the slug from which to start building the next page of results.
	NextPageToken string
}

ListConfig controls the behavior for listing StorageJobs.

type Select

type Select int

Select specifies which StorageJobs to list.

const (
	// All lists all StorageJobs and is the default.
	All Select = iota
	// Queued lists queued StorageJobs.
	Queued
	// Executing lists executing StorageJobs.
	Executing
	// Final lists final StorageJobs.
	Final
)

type StartedDeals

type StartedDeals struct {
	Cid          cid.Cid
	ProposalCids []cid.Cid
}

StartedDeals describe deals that are currently waiting to have a final status.

type Stats

type Stats struct {
	TotalQueued    int
	TotalExecuting int
}

Stats return metrics about current job queues.

type Store

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

Store is a Datastore implementation of JobStore, which saves state of scheduler Jobs.

func New

func New(ds datastore.TxnDatastore) (*Store, error)

New returns a new JobStore backed by the Datastore.

func (*Store) AddStartedDeals

func (s *Store) AddStartedDeals(iid ffs.APIID, c cid.Cid, proposals []cid.Cid) error

AddStartedDeals is a temporal storage solution of deals that are started are being watched. It serves as a recovery point to reattach to fired deals when the scheduler was abruptly interrupted.

func (*Store) CancelQueued

func (s *Store) CancelQueued(jid ffs.JobID) (bool, error)

CancelQueued cancels a job if it's in Queued status. If the Job isn't Queued, the call is a noop. If the Job doesn't exist it returns ErrNotFound.

func (*Store) Close

func (s *Store) Close() error

Close closes the Store, unregistering any subscribed watchers.

func (*Store) Dequeue

func (s *Store) Dequeue(iid ffs.APIID) (*ffs.StorageJob, error)

Dequeue dequeues a Job which doesn't have have another Executing Job for the same Cid. Saying it differently, it's safe to execute. The returned job Status is automatically changed to Executing. If an instance id is provided, only a job for that instance id will be dequeued. If no jobs are available to dequeue it returns a nil *ffs.Job and no-error.

func (*Store) Enqueue

func (s *Store) Enqueue(j ffs.StorageJob) error

Enqueue queues a new Job. If other Job for the same Cid is in Queued status, it will be automatically marked as Canceled.

func (*Store) Finalize

func (s *Store) Finalize(jid ffs.JobID, st ffs.JobStatus, jobError error, dealErrors []ffs.DealError) error

Finalize sets a Job status to a final state, i.e. Success or Failed, with a list of Deal errors occurred during job execution.

func (*Store) Get

func (s *Store) Get(jid ffs.JobID) (ffs.StorageJob, error)

Get returns the current state of Job. If doesn't exist, returns ErrNotFound.

func (*Store) GetExecutingJob

func (s *Store) GetExecutingJob(iid ffs.APIID, c cid.Cid) *ffs.JobID

GetExecutingJob returns a JobID that is currently executing for data with cid c in iid. If there's not such job, it returns nil.

func (*Store) GetExecutingJobIDs

func (s *Store) GetExecutingJobIDs() []ffs.JobID

GetExecutingJobIDs returns the JobIDs of all Jobs in Executing status.

func (*Store) GetStartedDeals

func (s *Store) GetStartedDeals(iid ffs.APIID, c cid.Cid) ([]cid.Cid, error)

GetStartedDeals gets all stored started deals from Cid for an APIID. If no started deals are present, an empty slice is returned.

func (*Store) GetStats

func (s *Store) GetStats() Stats

GetStats return the current Stats for storage jobs.

func (*Store) List

func (s *Store) List(config ListConfig) ([]ffs.StorageJob, bool, string, error)

List lists StorageJobs according to the provided ListConfig.

func (*Store) MonitorJob

func (s *Store) MonitorJob(j ffs.StorageJob) chan deals.StorageDealInfo

MonitorJob returns a channel that can be passed into the deal monitoring process.

func (*Store) RemoveStartedDeals

func (s *Store) RemoveStartedDeals(iid ffs.APIID, c cid.Cid) error

RemoveStartedDeals removes all started deals from Cid.

func (*Store) Watch

func (s *Store) Watch(ctx context.Context, c chan<- ffs.StorageJob, iid ffs.APIID) error

Watch subscribes to Job changes from a specified Api instance.