state

Documentation

Index

• Constants
• func WipeAndReset(dbConn DBConnector, migrator Migrator) error
• type DBConnector
  • func NewDBConnector(driver string, conn string) (DBConnector, error)
• type Migrator
  • func NewMigrator(driver string) (Migrator, error)
• type RecordUpdateStatus
• type SqlDatastore
  • func NewSqlDatastore(connector DBConnector, queries sqlds.Queries) *SqlDatastore
  • func (sds *SqlDatastore) Batch(ctx context.Context) (datastore.Batch, error)
  • func (sds *SqlDatastore) Close() error
  • func (sds *SqlDatastore) Delete(ctx context.Context, key datastore.Key) error
  • func (sds *SqlDatastore) Get(ctx context.Context, key datastore.Key) (value []byte, err error)
  • func (sds *SqlDatastore) GetSize(ctx context.Context, key datastore.Key) (size int, err error)
  • func (sds *SqlDatastore) Has(ctx context.Context, key datastore.Key) (exists bool, err error)
  • func (sds *SqlDatastore) Put(ctx context.Context, key datastore.Key, value []byte) error
  • func (sds *SqlDatastore) Query(ctx context.Context, q query.Query) (query.Results, error)
  • func (sds *SqlDatastore) Sync(ctx context.Context, prefix datastore.Key) error
• type State
  • func NewStateDB(ctx context.Context, dbConn DBConnector, migrator Migrator, logfile string, ...) (State, error)
• type Store

Constants

const ErrNoDeleteInProgressTasks = errorString("can only delete tasks that are not-started or tasks that are scheduled")

const ErrNotAssigned = errorString("tasks must be acquired through pop task")

const ErrTaskNotFound = errorString("task does not exist")

const ErrWrongWorker = errorString("task already acquired by other worker")

Functions

func WipeAndReset

func WipeAndReset(dbConn DBConnector, migrator Migrator) error

Types

type DBConnector

type DBConnector interface {
	Connect() error
	Name() string
	RetryableError(error) bool
	SqlDB() *sql.DB
}

DBConnector provides an interface for working with the underlying DB implementations

func NewDBConnector

func NewDBConnector(driver string, conn string) (DBConnector, error)

type Migrator

type Migrator func(*sql.DB) error

func NewMigrator

func NewMigrator(driver string) (Migrator, error)

type RecordUpdateStatus

type RecordUpdateStatus int

const (
	LATEST_UPDATE     RecordUpdateStatus = 1
	PREVIOUS_UPDATE   RecordUpdateStatus = 2
	UNATTACHED_RECORD RecordUpdateStatus = 10
	ATTACHED_RECORD   RecordUpdateStatus = 11
)

type SqlDatastore

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

SqlDatastore wraps a datastore.Batching with connection retry implemented in state.DBConnector. We need this because state.DBConnector seems to exist because db connections may drop and timeout during prolonged periods of inactivity, for example. The state.DBConnector is a dealbot construct; therefore, here we wrap a datastore to dynamically reconnect when the DB connection is closed.

func NewSqlDatastore

func NewSqlDatastore(connector DBConnector, queries sqlds.Queries) *SqlDatastore

NewSqlDatastore instantiates a new legs datastore backed by DB.

func (*SqlDatastore) Batch

func (sds *SqlDatastore) Batch(ctx context.Context) (datastore.Batch, error)

func (*SqlDatastore) Close

func (sds *SqlDatastore) Close() error

func (*SqlDatastore) Delete

func (sds *SqlDatastore) Delete(ctx context.Context, key datastore.Key) error

func (*SqlDatastore) Get

func (sds *SqlDatastore) Get(ctx context.Context, key datastore.Key) (value []byte, err error)

func (*SqlDatastore) GetSize

func (sds *SqlDatastore) GetSize(ctx context.Context, key datastore.Key) (size int, err error)

func (*SqlDatastore) Has

func (sds *SqlDatastore) Has(ctx context.Context, key datastore.Key) (exists bool, err error)

func (*SqlDatastore) Put

func (sds *SqlDatastore) Put(ctx context.Context, key datastore.Key, value []byte) error

func (*SqlDatastore) Query

func (sds *SqlDatastore) Query(ctx context.Context, q query.Query) (query.Results, error)

func (*SqlDatastore) Sync

func (sds *SqlDatastore) Sync(ctx context.Context, prefix datastore.Key) error

type State

type State interface {
	AssignTask(ctx context.Context, req tasks.PopTask) (tasks.Task, error)
	Get(ctx context.Context, uuid string) (tasks.Task, error)
	GetAll(ctx context.Context) ([]tasks.Task, error)
	GetHead(ctx context.Context, walkback int) (tasks.RecordUpdate, error)
	Update(ctx context.Context, uuid string, req tasks.UpdateTask) (tasks.Task, error)
	NewStorageTask(ctx context.Context, storageTask tasks.StorageTask) (tasks.Task, error)
	NewRetrievalTask(ctx context.Context, retrievalTask tasks.RetrievalTask) (tasks.Task, error)
	DrainWorker(ctx context.Context, worker string) error
	UndrainWorker(ctx context.Context, worker string) error
	PublishRecordsFrom(ctx context.Context, worker string) (cid.Cid, error)
	ResetWorkerTasks(ctx context.Context, worker string) error
	Delete(ctx context.Context, uuid string) error
	Store(ctx context.Context) Store
}

State provides an interface for persistence.

func NewStateDB

func NewStateDB(ctx context.Context, dbConn DBConnector, migrator Migrator, logfile string, identity crypto.PrivKey, recorder metrics.MetricsRecorder) (State, error)

NewStateDB creates a state instance with a given driver and identity

type Store

type Store interface {
	Head() (cid.Cid, error)
	Has(context.Context, string) (bool, error)
	Get(context.Context, string) ([]byte, error)
	Put(ctx context.Context, key string, content []byte) error
}