shardservice

type Config struct {
	// ServiceID service id
	ServiceID string `toml:"-"`
	// RPC rpc config
	RPC morpc.Config `toml:"-"`

	// Enable enable shard service
	Enable bool `toml:"enable"`
	// ListenAddress shard service listen address for receiving lock requests
	ListenAddress string `toml:"listen-address"`
	// ServiceAddress service address for communication, if this address is not set, use
	// ListenAddress as the communication address.
	ServiceAddress string `toml:"service-address"`
	// MaxScheduleTables how many tables's shards can be scheduled by balancer at once.
	MaxScheduleTables int `toml:"max-schedule-tables"`
	// FreezeCNTimeout max freeze time for CN can be scheduled by balancer again.
	FreezeCNTimeout toml.Duration `toml:"max-cn-freeze-time"`
	// ScheduleDuration how often to schedule shards on CNs.
	ScheduleDuration toml.Duration `toml:"schedule-duration"`
	// SelectCNLabel label name for tenant to select CNs.
	SelectCNLabel string `toml:"cn-select-label-name"`
	// HeartbeatDuration how often to send heartbeat to shard server.
	HeartbeatDuration toml.Duration `toml:"heartbeat-duration"`
	// CheckChangedDuration how often to check deleted tables or sharding changed in CN.
	CheckChangedDuration toml.Duration `toml:"check-changed-duration"`
}

type Env interface {
	HasCN(serviceID string) bool
	Available(accountID uint64, cn string) bool
	Draining(cn string) bool

	UpdateState(cn string, state metadata.WorkState)
}

type Option func(*service)

type ReadFunc func(
	ctx context.Context,
	shard pb.TableShard,
	engine engine.Engine,
	param pb.ReadParam,
	ts timestamp.Timestamp,
	buffer *morpc.Buffer,
) ([]byte, error)

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

type ReadRequest struct {
	TableID uint64
	Method  int
	Param   pb.ReadParam
	Apply   func([]byte)
}

type ServerOption func(*server)

type ShardServer interface {
	// Close close the shard server
	Close() error
}

type ShardService interface {
	// Config returns the configuration of the shard service.
	Config() Config
	// GetStorage returns the storage of the shard service.
	GetStorage() ShardStorage
	// Read read data from shards.
	Read(ctx context.Context, req ReadRequest, opts ReadOptions) error
	// HasLocalReplica returns whether the shard has a local replica.
	HasLocalReplica(tableID, shardID uint64) (bool, error)
	// HasAllLocalReplicas returns whether all shards of the table have local replicas.
	HasAllLocalReplicas(tableID uint64) (bool, error)
	// GetShardInfo returns the metadata of the shards corresponding to the table.
	GetShardInfo(table uint64) (shardTableID uint64, policy pb.Policy, isShardTable bool, err error)
	// GetShardAllocated returns the allocation of the shards corresponding to the table.
	GetTableShards(table uint64) (pb.ShardsMetadata, []pb.TableShard, error)
	// Create creates table shards metadata in current txn. And create shard
	// binds after txn committed asynchronously. Nothing happened if txn aborted.
	//
	// ShardBalancer will allocate CN after TableShardBind created.
	Create(ctx context.Context, table uint64, txnOp client.TxnOperator) error
	// Delete deletes table shards metadata in current txn. Table shards need
	// to be deleted if table deleted. Nothing happened if txn aborted.
	Delete(ctx context.Context, table uint64, txnOp client.TxnOperator) error
	// ReplicaCount returns the number of running replicas on current cn.
	ReplicaCount() int64
	// TableReplicaCount returns the number of running replicas of the special table on current cn.
	TableReplicaCount(tableID uint64) int64
	// Close close the service
	Close() error
}

type ShardStorage interface {
	// Get returns the latest metadata of the shards corresponding to the table.
	Get(table uint64) (uint64, pb.ShardsMetadata, error)
	// GetChanged returns the table ids of the shards that have been changed.
	GetChanged(tables map[uint64]uint32, applyDeleted func(uint64), applyChanged func(uint64)) error
	// Create creates the metadata for the sharding corresponding to the table with the given
	// transaction.
	Create(ctx context.Context, table uint64, txnOp client.TxnOperator) (bool, error)
	// Create delete the metadata for the sharding corresponding to the table with the given
	// transaction.
	Delete(ctx context.Context, table uint64, txnOp client.TxnOperator) (bool, error)
	// Unsubscribe unsubscribes the log tail of the tables.
	Unsubscribe(tables ...uint64) error
	// WaitLogAppliedAt wait until the log tail corresponding to ts has been fully consumed.
	// Ensure that subsequent reads have full log tail data.
	WaitLogAppliedAt(ctx context.Context, ts timestamp.Timestamp) error
	// Read read data with the given timestamp
	Read(ctx context.Context, shard pb.TableShard, method int, param pb.ReadParam, ts timestamp.Timestamp, buffer *morpc.Buffer) ([]byte, error)
}