replication

type Cache ¶ added in v0.25.0

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

Cache is an in-memory implementation of a cache for storing hydrated replication messages. Messages can come out of order as long as their task ID is higher than last acknowledged message. Out of order is expected as different source clusters will share hydrated replication messages.

Cache utilizes heap to keep replication messages in order. This is needed for efficient acknowledgements in O(log N).

Cache capacity can be increased dynamically. Decrease will require a restart, as new tasks will not be accepted, but memory will not be reclaimed either.

Cache methods are thread safe. It is expected to have writers and readers from different go routines.

func NewCache(capacity dynamicconfig.IntPropertyFn) *Cache

func (c *Cache) Ack(level int64)

func (c *Cache) Get(taskID int64) *types.ReplicationTask

func (c *Cache) Put(task *types.ReplicationTask) error

func (c *Cache) Size() int

type DLQHandler ¶

type DLQHandler interface {
	common.Daemon

	GetMessageCount(
		ctx context.Context,
		forceFetch bool,
	) (map[string]int64, error)
	ReadMessages(
		ctx context.Context,
		sourceCluster string,
		lastMessageID int64,
		pageSize int,
		pageToken []byte,
	) ([]*types.ReplicationTask, []*types.ReplicationTaskInfo, []byte, error)
	PurgeMessages(
		ctx context.Context,
		sourceCluster string,
		lastMessageID int64,
	) error
	MergeMessages(
		ctx context.Context,
		sourceCluster string,
		lastMessageID int64,
		pageSize int,
		pageToken []byte,
	) ([]byte, error)
}

DLQHandler is the interface handles replication DLQ messages

func NewDLQHandler(
	shard shard.Context,
	taskExecutors map[string]TaskExecutor,
) DLQHandler

type DynamicTaskReader ¶ added in v0.25.0

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

DynamicTaskReader will read replication tasks from database using dynamic batch sizing depending on replication lag.

func NewDynamicTaskReader(
	shardID int,
	executionManager persistence.ExecutionManager,
	timeSource clock.TimeSource,
	config *config.Config,
) *DynamicTaskReader

func (r *DynamicTaskReader) Read(ctx context.Context, readLevel int64, maxReadLevel int64) ([]*persistence.ReplicationTaskInfo, bool, error)

type MetricsEmitterImpl ¶ added in v0.25.0

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

MetricsEmitterImpl is responsible for emitting source side replication metrics occasionally.

func NewMetricsEmitter(
	shardID int,
	shardData metricsEmitterShardData,
	reader taskReader,
	metricsClient metrics.Client,
) *MetricsEmitterImpl

func (m *MetricsEmitterImpl) Start()

func (m *MetricsEmitterImpl) Stop()

type TaskAckManager ¶ added in v0.15.0

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

TaskAckManager is the ack manager for replication tasks

func NewTaskAckManager(
	shardID int,
	ackLevels ackLevelStore,
	metricsClient metrics.Client,
	logger log.Logger,
	reader taskReader,
	store *TaskStore,
) TaskAckManager

func (t *TaskAckManager) GetTasks(ctx context.Context, pollingCluster string, lastReadTaskID int64) (*types.ReplicationMessages, error)

type TaskExecutor ¶

type TaskExecutor interface {
	// contains filtered or unexported methods
}

TaskExecutor is the executor for replication task

func NewTaskExecutor(
	shard shard.Context,
	domainCache cache.DomainCache,
	historyResender ndc.HistoryResender,
	historyEngine engine.Engine,
	metricsClient metrics.Client,
	logger log.Logger,
) TaskExecutor

type TaskFetcher ¶

type TaskFetcher interface {
	common.Daemon

	GetSourceCluster() string
	GetRequestChan() chan<- *request
	GetRateLimiter() *quotas.DynamicRateLimiter
}

TaskFetcher is responsible for fetching replication messages from remote DC.

type TaskFetchers ¶

type TaskFetchers interface {
	common.Daemon

	GetFetchers() []TaskFetcher
}

TaskFetchers is a group of fetchers, one per source DC.

func NewTaskFetchers(
	logger log.Logger,
	config *config.Config,
	clusterMetadata cluster.Metadata,
	clientBean client.Bean,
) TaskFetchers

type TaskHydrator ¶ added in v0.25.0

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

TaskHydrator will enrich replication task with additional information from mutable state and history events. Mutable state and history providers can be either in-memory or persistence based implementations; depending whether we have available data already or need to load it.

func NewDeferredTaskHydrator(shardID int, historyManager persistence.HistoryManager, executionCache *execution.Cache, domains domainCache) TaskHydrator

func NewImmediateTaskHydrator(isRunning bool, vh *persistence.VersionHistories, activities map[int64]*persistence.ActivityInfo, blob, nextBlob *persistence.DataBlob) TaskHydrator

func (h TaskHydrator) Hydrate(ctx context.Context, task persistence.ReplicationTaskInfo) (retTask *types.ReplicationTask, retErr error)

type TaskProcessor ¶

type TaskProcessor interface {
	common.Daemon
}

TaskProcessor is responsible for processing replication tasks for a shard.

func NewTaskProcessor(
	shard shard.Context,
	historyEngine engine.Engine,
	config *config.Config,
	metricsClient metrics.Client,
	taskFetcher TaskFetcher,
	taskExecutor TaskExecutor,
) TaskProcessor

type TaskStore ¶ added in v0.25.0

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

TaskStore is a component that hydrates and caches replication messages so that they can be reused across several polling source clusters. It also exposes public Put method. This allows pre-store already hydrated messages at the end of successful transaction, saving a DB call to fetch history events.

TaskStore uses a separate cache per each source cluster allowing messages to be fetched at different rates. Once a cache becomes full it will not accept further messages for that cluster. Later those messages be fetched from DB and hydrated again. A cache stores only a pointer to the message. It is hydrates once and shared across caches. Cluster acknowledging the message will remove it from that corresponding cache. Once all clusters acknowledge it, no more references will be held, and GC will eventually pick it up.

func NewTaskStore(
	config *config.Config,
	clusterMetadata cluster.Metadata,
	domains domainCache,
	metricsClient metrics.Client,
	logger log.Logger,
	hydrator taskHydrator,
) *TaskStore

func (m *TaskStore) Ack(cluster string, lastTaskID int64) error

func (m *TaskStore) Get(ctx context.Context, cluster string, info persistence.ReplicationTaskInfo) (*types.ReplicationTask, error)

func (m *TaskStore) Put(task *types.ReplicationTask)