redis

README

Redis backend

Instances and their state

Instances and their state (started_at, completed_at etc.) are stored as JSON blobs under the instances-{instanceID} keys.

History and pending events

Events are stored in streams per workflow instance under the events-{instanceID} key. We maintain a cursor in the instance state, that indicates the last event that has been executed. Every event after that in the stream, is a pending event and will be returned to the worker in the next workflow task.

Timer events

Timer events are stored in a sorted set (ZSET). Whenever a worker checks for a new workflow instance task, the sorted set is checked to see if any of the pending timer events is ready yet. If it is, it's added to the pending events before those are returned for pending workflow tasks.

Task queues

We need queues for activities and workflow instances. In both cases, we have tasks being enqueued, workers polling for works, and we have to guarantee that every task is eventually processed. So if a worker has dequeued a task and crashed, for example, eventually we need another worker to pick up the task and finish it.

Task queues are implemented using Redis STREAMs. In addition for queues where we only want a single instance of a task to be in the queue, we maintain an additional SET.

Alternatives considered

Option 1 - ZSET:

Store keys for queue items in a ZSET. The score for the sorted set is the timestamp when the task is unlocked. For new tasks the SCORE is the current timestamp.

Workers make a ZRANGE query to that sorted set, looking for tasks where the SCORE is in -inf, now). That will get tasksß where the unlock timestamp is in the past. Either because they have just been queued, or the lock has expired.

Once a worker picks up a task, the score is updated to now + lock_timeout. Query and update are done in a transaction with a WATCH on the queue key.

When a worker is done with a task, it removes it from the ZSET (ZREM).

Pro:

• No special handling for recovering crashed tasks, they'll automatically unlock

Con:

• Need for polling and cannot use any of the blocking redis commands
• WAIT with transaction, or a script required

Option 2 - LISTS

Use two LISTs, one for the pending queue, one for the processing queue. To enqueue a new task, LPUSH it onto the pending list. Also add an entry in a separate ZSET where the score is the unlock timestamp. Initially that timestamp will be the current timestamp. The LPUSH and ZADD are done in a transaction with a WATCH on the queue key, and retried if another client modified the queue in the mean time. Alternatively, the two operations can be done in a script.

For picking up tasks, we use a blocking BLMOVE .. RIGHT LEFT command to pick up the next available task from pending and move it to processing as a single atomic operation. Once picked up, the SCORE in the ZSET is adjusted to now + lock_timeout.

When a worker is done with a task, it removes it from the processing list (LREM), and the ZSET (ZREM).

To recover abandoned tasks, we periodically scan the processing list

Pro:

• Blocking call does not require constant polling

Con:

• Requires periodic scans of the processing list to find tasks that have been abandoned
• Picking up a task, adjusting its ZSET value and the periodic scan could run into race conditions

Documentation

Index

• func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (*redisBackend, error)
• type KeyInfo
• type RedisBackendOption
  • func WithAutoExpiration(expireFinishedRunsAfter time.Duration) RedisBackendOption
  • func WithAutoExpirationContinueAsNew(expireContinuedAsNewRunsAfter time.Duration) RedisBackendOption
  • func WithBackendOptions(opts ...backend.BackendOption) RedisBackendOption
  • func WithBlockTimeout(timeout time.Duration) RedisBackendOption
  • func WithKeyPrefix(prefix string) RedisBackendOption
• type RedisOptions
• type TaskItem

Functions

func NewRedisBackend

func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (*redisBackend, error)

Types

type KeyInfo

type KeyInfo struct {
	StreamKey string
	SetKey    string
}

type RedisBackendOption

type RedisBackendOption func(*RedisOptions)

func WithAutoExpiration

func WithAutoExpiration(expireFinishedRunsAfter time.Duration) RedisBackendOption

WithAutoExpiration sets the duration after which finished runs will expire from the data store. If set to 0 (default), runs will never expire and need to be manually removed.

func WithAutoExpirationContinueAsNew

func WithAutoExpirationContinueAsNew(expireContinuedAsNewRunsAfter time.Duration) RedisBackendOption

WithAutoExpirationContinueAsNew sets the duration after which runs that were completed with `ContinueAsNew` automatically expire. If set to 0 (default), the overall expiration setting set with `WithAutoExpiration` will be used.

func WithBackendOptions

func WithBackendOptions(opts ...backend.BackendOption) RedisBackendOption

func WithBlockTimeout

func WithBlockTimeout(timeout time.Duration) RedisBackendOption

WithBlockTimeout sets the timeout for blocking operations like dequeuing a workflow or activity task

func WithKeyPrefix

func WithKeyPrefix(prefix string) RedisBackendOption

WithKeyPrefix sets the prefix for all keys used in the Redis backend.

type RedisOptions

type RedisOptions struct {
	*backend.Options

	BlockTimeout time.Duration

	AutoExpiration              time.Duration
	AutoExpirationContinueAsNew time.Duration

	KeyPrefix string
}

type TaskItem

type TaskItem[T any] struct {
	// TaskID is the generated ID of the task item
	TaskID string

	// ID is the provided id
	ID string

	// Optional data stored with a task, needs to be serializable
	Data T
}