distrlock

README

Distributed Lock

A distributed lock can be used to ensure mutual exclusive access to a specific resource across different running applications.

To acquire a lock, the lock handler will try to create a new lock. This can only succeed if the no lock with the same key exists. One can overtake a lock (and refresh it) once a specified timeout is passed.

The lock handler can work on a table with only a Partition Key. Or on a table containing a Partition Key and Sort Key. In the later case, one can argue that we can lock a specific partition on the DynamoDB table.

Example

func foo(ctx context.Context, client *dynamodb.Client, tablename string, partitionKey string) error {
	lockHandler := distrlock.New(client, tablename, partitionKey, distrlock.WithTimeout(time.Second))
	
	lock, err := lockHandler.Lock(ctx, &types.AttributeValueMemberS{Value: "partitionToLock"})
	if err != nil {
        return err
    }
	
	defer lock.Release(ctx)
	
	// Lock is acquired. You can access specific data in a mutual exclusive way
}

Documentation

Index

• Variables
• func WithRefreshInterval(refreshInterval time.Duration) func(options *Options)
• func WithRefreshVariance(refreshVariance time.Duration) func(options *Options)
• func WithSortKey(sortKeyColumnName string) func(options *Options)
• func WithSortKeyValue(sortKeyValue types.AttributeValue) func(options *Options)
• func WithTimeout(timeout time.Duration) func(options *Options)
• type DynamodbClient
• type ErrDistrLock
  • func NewDistrLockError(msg string, err error) *ErrDistrLock
  • func (e *ErrDistrLock) Error() string
  • func (e *ErrDistrLock) Unwrap() error
• type IdGenerator
• type Lock
  • func (l *Lock) LockId() string
  • func (l *Lock) Refresh(ctx context.Context) error
  • func (l *Lock) Release(ctx context.Context) error
  • func (l *Lock) TransactionCondition() types.TransactWriteItem
  • func (l *Lock) TransactionWithRefresh() (types.TransactWriteItem, ...)
• type Options
• type RepositoryLockHandler
  • func New(client DynamodbClient, tableName string, partitionKeyName string, ...) *RepositoryLockHandler
  • func (h *RepositoryLockHandler) Lock(ctx context.Context, partition types.AttributeValue) (*Lock, error)
  • func (h *RepositoryLockHandler) TryLock(ctx context.Context, partition types.AttributeValue) (*Lock, bool, error)

Variables

var ErrLockUpdate = errors.New("lock update error")

var ErrTimeout = errors.New("timeout")

var SkString = &types.AttributeValueMemberS{Value: "##LOCK##"}

Functions

func WithRefreshInterval

func WithRefreshInterval(refreshInterval time.Duration) func(options *Options)

WithRefreshInterval Specifies a refresh interval that is used by the lock handler to poll if the lock is still active. Default value is 200ms

func WithRefreshVariance

func WithRefreshVariance(refreshVariance time.Duration) func(options *Options)

WithRefreshVariance Specifies a variance of the refresh interval that is used by the lock handler. Default value is 20ms.

func WithSortKey

func WithSortKey(sortKeyColumnName string) func(options *Options)

WithSortKey Specifies the sort key column if exists.

func WithSortKeyValue

func WithSortKeyValue(sortKeyValue types.AttributeValue) func(options *Options)

WithSortKeyValue Specifies the sort key value to use for a lock. This is required if the column hash a sort key that is not of they string. Note that the partition key + sort key must be unique within the table.

func WithTimeout

func WithTimeout(timeout time.Duration) func(options *Options)

WithTimeout Specifies a timeout value set in the distributed lock. Default value is 1 second.

Types

type DynamodbClient

type DynamodbClient interface {
	PutItem(ctx context.Context, params *dynamodb.PutItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.PutItemOutput, error)
	GetItem(ctx context.Context, params *dynamodb.GetItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.GetItemOutput, error)
	DeleteItem(ctx context.Context, params *dynamodb.DeleteItemInput, optFns ...func(options *dynamodb.Options)) (*dynamodb.DeleteItemOutput, error)
}

type ErrDistrLock

type ErrDistrLock struct {
	Msg string
	Err error
}

func NewDistrLockError

func NewDistrLockError(msg string, err error) *ErrDistrLock

func (*ErrDistrLock) Error

func (e *ErrDistrLock) Error() string

func (*ErrDistrLock) Unwrap

func (e *ErrDistrLock) Unwrap() error

type IdGenerator

type IdGenerator interface {
	ID() string
}

type Lock

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

func (*Lock) LockId

func (l *Lock) LockId() string

LockId returns the current id used by the lock

func (*Lock) Refresh

func (l *Lock) Refresh(ctx context.Context) error

Refresh updates the timeout of the current active lock

func (*Lock) Release

func (l *Lock) Release(ctx context.Context) error

Release remove the lock in the database

func (*Lock) TransactionCondition

func (l *Lock) TransactionCondition() types.TransactWriteItem

TransactionCondition returns a TransactWriteItem to validate if the lock is still active

func (*Lock) TransactionWithRefresh

func (l *Lock) TransactionWithRefresh() (types.TransactWriteItem, func(*dynamodb.TransactWriteItemsOutput, error) (*dynamodb.TransactWriteItemsOutput, error))

TransactionWithRefresh returns a TransactWriteItem to validate if the lock is still active and refresh the lock if successful Note the callback function returned as second argument should be called with the return types of the TransactWriteItems call

type Options

type Options struct {
	// SortKeyName if table has a sort key this value represent the sort key value
	SortKeyName *string

	// SortKeyDefault value
	SortKeyValue types.AttributeValue

	// Timeout stored in distributed lock
	Timeout *time.Duration

	// RefreshInterval average time between two lock poll request
	RefreshInterval *time.Duration

	// RefreshVariance variance of the refresh interval between two lock poll request
	RefreshVariance *time.Duration

	// IdGenerator a generator of unique IDs
	IdGenerator IdGenerator
}

type RepositoryLockHandler

type RepositoryLockHandler struct {
	Client           DynamodbClient
	TableName        string
	PartitionKeyName string
	SortKeyName      *string
	SortKeyValue     types.AttributeValue
	Timeout          time.Duration
	RefreshInterval  time.Duration
	RefreshVariance  time.Duration
	IdGenerator      IdGenerator
}

RepositoryLockHandler will handle locks distributed locks within a single dynamodb table The RepositoryLockHandler can create locks on hash tables and hash+range tables

func New

func New(client DynamodbClient, tableName string, partitionKeyName string, optFns ...func(options *Options)) *RepositoryLockHandler

New create a new initialized distributed lock.

func (*RepositoryLockHandler) Lock

func (h *RepositoryLockHandler) Lock(ctx context.Context, partition types.AttributeValue) (*Lock, error)

Lock tries to lock a specified partition. The method will return a new lock once it is able to create a lock. If it was unable to create a new lock it will try after a RefreshInterval. Polling will stop if the context is Done.

func (*RepositoryLockHandler) TryLock

func (h *RepositoryLockHandler) TryLock(ctx context.Context, partition types.AttributeValue) (*Lock, bool, error)

TryLock tries to lock a specified partition. If the handler was able to lock the partition a new lock will be returned. Additionally, the second return argument will be true If the handler was unable to lock the partition nil and false is returned as first arguments.