sjstore

Documentation

Index

• Variables
• type StartedDeals
• type Stats
• type Store
  • func New(ds datastore.Datastore) (*Store, error)
  • func (s *Store) AddStartedDeals(c cid.Cid, proposals []cid.Cid) error
  • func (s *Store) Close() error
  • func (s *Store) Dequeue() (*ffs.StorageJob, error)
  • func (s *Store) Enqueue(j ffs.StorageJob) error
  • func (s *Store) ExecutingJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • 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(c cid.Cid) *ffs.JobID
  • func (s *Store) GetExecutingJobIDs() []ffs.JobID
  • func (s *Store) GetStartedDeals(c cid.Cid) ([]cid.Cid, error)
  • func (s *Store) GetStats() Stats
  • func (s *Store) LatestFinalJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Store) LatestSuccessfulJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Store) MonitorJob(j ffs.StorageJob) chan deals.StorageDealInfo
  • func (s *Store) QueuedJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob
  • func (s *Store) RemoveStartedDeals(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 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.Datastore) (*Store, error)

New returns a new JobStore backed by the Datastore.

func (*Store) AddStartedDeals

func (s *Store) AddStartedDeals(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) Close

func (s *Store) Close() error

Close closes the Store, unregistering any subscribed watchers.

func (*Store) Dequeue

func (s *Store) Dequeue() (*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 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) ExecutingJobs

func (s *Store) ExecutingJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

ExecutingJobs returns executing jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

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(c cid.Cid) *ffs.JobID

GetExecutingJob returns a JobID that is currently executing for data with cid c. 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(c cid.Cid) ([]cid.Cid, error)

GetStartedDeals gets all stored started deals from Cid. 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) LatestFinalJobs

func (s *Store) LatestFinalJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

LatestFinalJobs returns the most recent finished jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

func (*Store) LatestSuccessfulJobs

func (s *Store) LatestSuccessfulJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

LatestSuccessfulJobs returns the most recent successful jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

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

func (s *Store) QueuedJobs(iid ffs.APIID, cids ...cid.Cid) []ffs.StorageJob

QueuedJobs returns queued jobs for the specified instance id and cids. If the instance id is ffs.EmptyInstanceID, data for all instances is returned. If no cids are provided, data for all data cids is returned.

func (*Store) RemoveStartedDeals

func (s *Store) RemoveStartedDeals(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.