diff --git a/pkg/distlock/doc.go b/pkg/distlock/doc.go index 7327fb56..08e30a93 100644 --- a/pkg/distlock/doc.go +++ b/pkg/distlock/doc.go @@ -1,5 +1,10 @@ // Package distlock provides distributed, automatically renewed leases across independent nodes. // +// Its quorum and lease semantics are based on the Redlock algorithm described at +// https://redis.io/docs/latest/develop/clients/patterns/distributed-locks/. The core package is independent of storage +// and network transport and does not require Redis. Applications that communicate with remote nodes over gRPC can use +// the grpc subpackage to expose and call node-local lease operations. +// // A Cluster must return every node in the lock group, including temporarily unavailable nodes, because every node // counts toward quorum. Changing the node set is unsafe if a new quorum can be disjoint from an earlier quorum while // leases acquired from the earlier node set may still be valid. Callers must stop protected work when the context diff --git a/pkg/distlock/locker.go b/pkg/distlock/locker.go new file mode 100644 index 00000000..c5df809a --- /dev/null +++ b/pkg/distlock/locker.go @@ -0,0 +1,393 @@ +package distlock + +import ( + "context" + "crypto/rand" + "errors" + "fmt" + "math" + "slices" + "sync" + "time" + + "github.com/cenkalti/backoff/v4" +) + +const ( + DefaultLeaseDuration = 10 * time.Second + DefaultClockDriftFactor = 0.01 + DefaultMaxNodeCallTimeout = 5 * time.Second +) + +var ( + // ErrLeaseLost is the cancellation cause when automatic renewal can no longer maintain a lease. + ErrLeaseLost = errors.New("distributed lease lost") + // ErrLeaseReleased is the cancellation cause of an explicitly released lease. + ErrLeaseReleased = errors.New("distributed lease released") +) + +// Config configures a Locker. +type Config struct { + // LeaseDuration is the TTL used for acquisitions and renewals. The default is DefaultLeaseDuration. + LeaseDuration time.Duration + // ClockDriftFactor is the fraction of LeaseDuration reserved for differences in clock rates between the Locker and + // nodes. The default is DefaultClockDriftFactor. + ClockDriftFactor float64 + // NodeCallTimeout sets the context timeout for an Acquire, Renew, or Release call to one node. + // The default is the smaller of DefaultMaxNodeCallTimeout and one third of the lease duration. + NodeCallTimeout time.Duration + // NewBackOff creates independent retry policies for acquisitions and renewal cycles. The default is an exponential + // backoff starting at 100ms and capped at 1s. A policy should not impose its own elapsed-time limit because the + // acquisition context and current lease validity already bound retries. + NewBackOff func() backoff.BackOff +} + +func (c Config) clockDrift() time.Duration { + return time.Duration(math.Ceil(float64(c.LeaseDuration) * c.ClockDriftFactor)) +} + +// Locker acquires automatically renewed distributed leases over a Cluster. +type Locker struct { + config Config + cluster Cluster +} + +// New creates a Locker over cluster. +func New(cluster Cluster, config Config) (*Locker, error) { + if cluster == nil { + return nil, fmt.Errorf("cluster is nil") + } + + if config.LeaseDuration == 0 { + config.LeaseDuration = DefaultLeaseDuration + } + if config.LeaseDuration < 0 { + return nil, fmt.Errorf("lease duration must be positive") + } + if config.ClockDriftFactor == 0 { + config.ClockDriftFactor = DefaultClockDriftFactor + } + if config.ClockDriftFactor <= 0 || config.ClockDriftFactor >= 1 { + return nil, fmt.Errorf("clock drift factor must be greater than 0 and less than 1") + } + if config.NodeCallTimeout == 0 { + config.NodeCallTimeout = min(DefaultMaxNodeCallTimeout, config.LeaseDuration/3) + } + if config.NodeCallTimeout < 0 { + return nil, fmt.Errorf("node call timeout must be positive") + } + if config.NewBackOff == nil { + config.NewBackOff = defaultBackOff + } + + return &Locker{config: config, cluster: cluster}, nil +} + +func defaultBackOff() backoff.BackOff { + return backoff.NewExponentialBackOff( + backoff.WithInitialInterval(100*time.Millisecond), + backoff.WithMaxInterval(time.Second), + backoff.WithMaxElapsedTime(0), + ) +} + +// Acquire waits until it acquires a lease for resource or ctx ends. +func (l *Locker) Acquire(ctx context.Context, resource string) (*Lease, error) { + if resource == "" { + return nil, fmt.Errorf("resource is empty") + } + if err := ctx.Err(); err != nil { + return nil, err + } + + nodes, err := l.cluster.Nodes(ctx) + if err != nil { + return nil, fmt.Errorf("get cluster nodes: %w", err) + } + if len(nodes) == 0 { + return nil, fmt.Errorf("cluster has no nodes") + } + nodes = slices.Clone(nodes) + + boff := backoff.WithContext(l.config.NewBackOff(), ctx) + var lease *Lease + err = backoff.Retry(func() error { + token, tokenErr := newOwnershipToken() + if tokenErr != nil { + return backoff.Permanent(fmt.Errorf("generate lease token: %w", tokenErr)) + } + + candidate := newLease(l, nodes, resource, token) + if err := candidate.acquire(ctx); err != nil { + return err + } + + lease = candidate + return nil + }, boff) + if err != nil { + return nil, fmt.Errorf("acquire distributed lease for %q: %w", resource, err) + } + + return lease, nil +} + +// newOwnershipToken generates a unique 128-bit random ownership token for a lease. +func newOwnershipToken() ([]byte, error) { + token := make([]byte, 16) + if _, err := rand.Read(token); err != nil { + return nil, err + } + return token, nil +} + +// Lease is an automatically renewed distributed lease. +type Lease struct { + config Config + nodes []Node + resource string + token []byte + + ctx context.Context + cancel context.CancelCauseFunc + // done is closed when the renewal goroutine exits. + done chan struct{} + + // operationMu prevents acquisition, renewal, and release operations for the lease from overlapping. + operationMu sync.Mutex + quorum int +} + +func newLease(locker *Locker, nodes []Node, resource string, token []byte) *Lease { + ctx, cancel := context.WithCancelCause(context.Background()) + return &Lease{ + config: locker.config, + nodes: nodes, + resource: resource, + token: token, + ctx: ctx, + cancel: cancel, + done: make(chan struct{}), + quorum: len(nodes)/2 + 1, + } +} + +// Context returns a context that is cancelled when the lease is lost or explicitly released. +func (l *Lease) Context() context.Context { + return l.ctx +} + +// Release stops automatic renewal and attempts to remove the lease from every node in its acquisition snapshot. +func (l *Lease) Release(ctx context.Context) error { + l.cancel(ErrLeaseReleased) + select { + case <-l.done: + case <-ctx.Done(): + return ctx.Err() + } + + if err := l.release(ctx); err != nil { + return fmt.Errorf("release distributed lease for %q: %w", l.resource, err) + } + return nil +} + +type nodeResult struct { + success bool + err error +} + +func collectNodeResults(results <-chan nodeResult) (successes int, err error) { + var errs []error + for result := range results { + if result.err != nil { + errs = append(errs, result.err) + } else if result.success { + successes++ + } + } + return successes, errors.Join(errs...) +} + +func (l *Lease) executeNodes(ctx context.Context, fn func(context.Context, Node) (bool, error)) <-chan nodeResult { + results := make(chan nodeResult, len(l.nodes)) + var wg sync.WaitGroup + for _, node := range l.nodes { + wg.Go(func() { + callCtx, cancel := context.WithTimeout(ctx, l.config.NodeCallTimeout) + defer cancel() + + success, err := fn(callCtx, node) + results <- nodeResult{success: success, err: err} + }) + } + go func() { + wg.Wait() + close(results) + }() + return results +} + +// acquire makes one attempt to obtain the lease from a quorum of nodes and starts renewal on success. +func (l *Lease) acquire(ctx context.Context) error { + startedAt := time.Now() + validUntil := startedAt.Add(l.config.LeaseDuration - l.config.clockDrift()) + resultCh := make(chan error, 1) + // Aggregate asynchronously so acquire can return at quorum while this goroutine drains the remaining results and + // holds operationMu until every node call has finished. + go func() { + l.operationMu.Lock() + defer l.operationMu.Unlock() + + // Use the lease context so cancelling the context passed to Acquire after it succeeds does not stop node calls + // still pending after quorum. On failure, acquire cancels the lease context below. The validity deadline and + // per-node NodeCallTimeout bound these calls. + operationCtx, cancel := context.WithDeadline(l.ctx, validUntil) + defer cancel() + results := l.executeNodes(operationCtx, func(ctx context.Context, node Node) (bool, error) { + return node.Acquire(ctx, l.resource, l.token, l.config.LeaseDuration) + }) + + successes := 0 + errs := make([]error, 0, len(l.nodes)) + reported := false + for result := range results { + if result.err != nil { + errs = append(errs, result.err) + } else if result.success { + successes++ + } + if !reported && successes >= l.quorum { + resultCh <- nil + reported = true + } + } + if !reported { + quorumErr := fmt.Errorf("lease acquired on %d of %d nodes, need at least %d", + successes, len(l.nodes), l.quorum) + resultCh <- errors.Join(quorumErr, errors.Join(errs...)) + } + }() + + var acquireErr error + select { + case acquireErr = <-resultCh: + case <-ctx.Done(): + acquireErr = ctx.Err() + } + + if acquireErr == nil { + if !time.Now().Before(validUntil) { + acquireErr = fmt.Errorf("lease validity expired during acquisition") + } else { + go l.runRenew(validUntil) + return nil + } + } + + // Cancel any node Acquire calls still in progress. release waits for them to finish before removing partial leases, + // so no node can create this lease after cleanup has run. + l.cancel(acquireErr) + _ = l.release(context.WithoutCancel(ctx)) + return acquireErr +} + +// release waits for any in-progress lease operation, removes the lease from every node, and returns any errors. +func (l *Lease) release(ctx context.Context) error { + l.operationMu.Lock() + defer l.operationMu.Unlock() + if err := ctx.Err(); err != nil { + return err + } + + results := l.executeNodes(ctx, func(ctx context.Context, node Node) (bool, error) { + return node.Release(ctx, l.resource, l.token) + }) + _, err := collectNodeResults(results) + return err +} + +// runRenew periodically renews the lease until it is released or lost. +func (l *Lease) runRenew(validUntil time.Time) { + defer close(l.done) + + for { + remaining := time.Until(validUntil) + if remaining <= 0 { + break + } + + // Start renewal with two thirds of the current validity remaining + // to leave time for slow node calls and retries. + timer := time.NewTimer(remaining / 3) + select { + case <-l.ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + + renewedUntil, err := l.renew(validUntil) + if err != nil { + if l.ctx.Err() != nil { + return + } + break + } + validUntil = renewedUntil + } + + l.cancel(ErrLeaseLost) + _ = l.release(context.Background()) +} + +// renew retries node renewals until a quorum succeeds, the configured backoff stops, or the current lease validity +// ends. It returns the new validity deadline after reaching quorum. +func (l *Lease) renew(currentValidUntil time.Time) (time.Time, error) { + ctx, cancel := context.WithDeadline(l.ctx, currentValidUntil) + defer cancel() + + var validUntil time.Time + resultCh := make(chan error, 1) + // Coordinate in the background so renew can report expiry even if a node call does not return after cancellation. + // Keep operationMu held until every call finishes so cleanup cannot race a pending renewal. + go func() { + l.operationMu.Lock() + defer l.operationMu.Unlock() + + if err := ctx.Err(); err != nil { + resultCh <- err + return + } + + boff := backoff.WithContext(l.config.NewBackOff(), ctx) + resultCh <- backoff.Retry(func() error { + startedAt := time.Now() + results := l.executeNodes(ctx, func(ctx context.Context, node Node) (bool, error) { + return node.Renew(ctx, l.resource, l.token, l.config.LeaseDuration) + }) + successes, nodeErr := collectNodeResults(results) + if ctx.Err() != nil { + return ctx.Err() + } + if !time.Now().Before(currentValidUntil) { + return backoff.Permanent(fmt.Errorf("renewal attempt took longer than its validity window")) + } + + if successes >= l.quorum { + validUntil = startedAt.Add(l.config.LeaseDuration - l.config.clockDrift()) + return nil + } + quorumErr := fmt.Errorf("lease renewed on %d of %d nodes, need at least %d", + successes, len(l.nodes), l.quorum) + return errors.Join(quorumErr, nodeErr) + }, boff) + }() + + select { + case err := <-resultCh: + return validUntil, err + case <-ctx.Done(): + return time.Time{}, ctx.Err() + } +} diff --git a/pkg/distlock/locker_test.go b/pkg/distlock/locker_test.go new file mode 100644 index 00000000..916673dc --- /dev/null +++ b/pkg/distlock/locker_test.go @@ -0,0 +1,445 @@ +package distlock_test + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/psviderski/uncloud/pkg/distlock" + "github.com/stretchr/testify/require" +) + +var errMemoryNodeUnavailable = errors.New("memory node unavailable") + +type memoryCluster struct { + nodes []*memoryNode +} + +func newMemoryCluster(size int) *memoryCluster { + cluster := &memoryCluster{nodes: make([]*memoryNode, size)} + for i := range cluster.nodes { + cluster.nodes[i] = &memoryNode{store: distlock.NewMemoryStore(), available: true} + } + return cluster +} + +func (c *memoryCluster) Nodes(ctx context.Context) ([]distlock.Node, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + + nodes := make([]distlock.Node, len(c.nodes)) + for i, node := range c.nodes { + nodes[i] = node + } + return nodes, nil +} + +type memoryNode struct { + mu sync.Mutex + store distlock.Store + available bool + unavailableObserved chan struct{} +} + +func (n *memoryNode) setAvailable(available bool) { + if !available { + n.makeUnavailable() + return + } + + n.mu.Lock() + defer n.mu.Unlock() + n.available = true +} + +func (n *memoryNode) makeUnavailable() <-chan struct{} { + n.mu.Lock() + defer n.mu.Unlock() + n.available = false + n.unavailableObserved = make(chan struct{}) + return n.unavailableObserved +} + +func (n *memoryNode) restart() { + n.mu.Lock() + defer n.mu.Unlock() + n.store = distlock.NewMemoryStore() +} + +func (n *memoryNode) currentStore(ctx context.Context) (distlock.Store, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + + n.mu.Lock() + defer n.mu.Unlock() + if !n.available { + if n.unavailableObserved != nil { + close(n.unavailableObserved) + n.unavailableObserved = nil + } + return nil, errMemoryNodeUnavailable + } + return n.store, nil +} + +func (n *memoryNode) Acquire( + ctx context.Context, resource string, token []byte, ttl time.Duration, +) (bool, error) { + store, err := n.currentStore(ctx) + if err != nil { + return false, err + } + return store.Acquire(ctx, resource, token, ttl) +} + +func (n *memoryNode) Renew( + ctx context.Context, resource string, token []byte, ttl time.Duration, +) (bool, error) { + store, err := n.currentStore(ctx) + if err != nil { + return false, err + } + return store.Renew(ctx, resource, token, ttl) +} + +func (n *memoryNode) Release(ctx context.Context, resource string, token []byte) (bool, error) { + store, err := n.currentStore(ctx) + if err != nil { + return false, err + } + return store.Release(ctx, resource, token) +} + +func retryBackOff() backoff.BackOff { + return backoff.NewConstantBackOff(5 * time.Millisecond) +} + +func oneAttemptBackOff() backoff.BackOff { + return &backoff.StopBackOff{} +} + +func newTestLocker( + t *testing.T, cluster distlock.Cluster, leaseDuration time.Duration, newBackOff func() backoff.BackOff, +) *distlock.Locker { + t.Helper() + locker, err := distlock.New(cluster, distlock.Config{ + LeaseDuration: leaseDuration, + NodeCallTimeout: leaseDuration / 3, + NewBackOff: newBackOff, + }) + require.NoError(t, err) + return locker +} + +func acquireLease(t *testing.T, locker *distlock.Locker, resource string) *distlock.Lease { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + lease, err := locker.Acquire(ctx, resource) + require.NoError(t, err) + return lease +} + +func releaseLease(t *testing.T, lease *distlock.Lease) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + require.NoError(t, lease.Release(ctx)) +} + +func requireNoReceive[T any](t *testing.T, ch <-chan T, timeout time.Duration, message string) { + t.Helper() + select { + case <-ch: + require.FailNow(t, message) + case <-time.After(timeout): + } +} + +func requireReceive[T any](t *testing.T, ch <-chan T, timeout time.Duration, message string) T { + t.Helper() + select { + case value := <-ch: + return value + case <-time.After(timeout): + require.FailNow(t, message) + var zero T + return zero + } +} + +func requireContextActive(t *testing.T, ctx context.Context, message string) { + t.Helper() + select { + case <-ctx.Done(): + require.FailNow(t, message, "cause: %v", context.Cause(ctx)) + default: + } +} + +func requireSignal(t *testing.T, ch <-chan struct{}, timeout time.Duration, message string) { + t.Helper() + select { + case <-ch: + case <-time.After(timeout): + require.FailNow(t, message) + } +} + +func TestLockerAcquireQuorumBoundaries(t *testing.T) { + tests := []struct { + name string + nodes int + unavailable int + wantAcquire bool + }{ + {name: "two nodes at quorum", nodes: 2, wantAcquire: true}, + {name: "two nodes below quorum", nodes: 2, unavailable: 1, wantAcquire: false}, + {name: "three nodes at quorum", nodes: 3, unavailable: 1, wantAcquire: true}, + {name: "three nodes below quorum", nodes: 3, unavailable: 2, wantAcquire: false}, + {name: "four nodes at quorum", nodes: 4, unavailable: 1, wantAcquire: true}, + {name: "four nodes below quorum", nodes: 4, unavailable: 2, wantAcquire: false}, + {name: "five nodes at quorum", nodes: 5, unavailable: 2, wantAcquire: true}, + {name: "five nodes below quorum", nodes: 5, unavailable: 3, wantAcquire: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cluster := newMemoryCluster(tt.nodes) + unavailableObserved := make([]<-chan struct{}, 0, tt.unavailable) + for i := range tt.unavailable { + unavailableObserved = append(unavailableObserved, cluster.nodes[tt.nodes-1-i].makeUnavailable()) + } + locker := newTestLocker(t, cluster, 300*time.Millisecond, oneAttemptBackOff) + + acquireCtx, cancelAcquire := context.WithTimeout(context.Background(), 2*time.Second) + lease, err := locker.Acquire(acquireCtx, "resource") + cancelAcquire() + for _, observed := range unavailableObserved { + requireSignal(t, observed, time.Second, "unavailable node did not receive acquisition") + } + for i := range tt.unavailable { + cluster.nodes[tt.nodes-1-i].setAvailable(true) + } + + if tt.wantAcquire { + require.NoError(t, err) + require.NotNil(t, lease) + releaseLease(t, lease) + } else { + if lease != nil { + releaseLease(t, lease) + } + require.Error(t, err) + require.Nil(t, lease) + } + }) + } +} + +func TestLockerSingleNodeLifecycle(t *testing.T) { + const leaseDuration = 300 * time.Millisecond + cluster := newMemoryCluster(1) + locker := newTestLocker(t, cluster, leaseDuration, oneAttemptBackOff) + + lease := acquireLease(t, locker, "resource") + defer func() { + if lease.Context().Err() != nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = lease.Release(ctx) + }() + + time.Sleep(leaseDuration + 100*time.Millisecond) + requireContextActive(t, lease.Context(), "single-node lease was not renewed") + releaseLease(t, lease) + + secondLease := acquireLease(t, locker, "resource") + releaseLease(t, secondLease) +} + +func TestLockerFailedAcquireCleansUpPartialLease(t *testing.T) { + cluster := newMemoryCluster(3) + cluster.nodes[1].setAvailable(false) + cluster.nodes[2].setAvailable(false) + locker := newTestLocker(t, cluster, 300*time.Millisecond, oneAttemptBackOff) + + lease, err := locker.Acquire(context.Background(), "resource") + require.Error(t, err) + require.Nil(t, lease) + + cluster.nodes[1].setAvailable(true) + cluster.nodes[2].setAvailable(true) + lease = acquireLease(t, locker, "resource") + releaseLease(t, lease) +} + +func TestLockerAcquiresIndependentResourcesConcurrently(t *testing.T) { + cluster := newMemoryCluster(3) + locker := newTestLocker(t, cluster, 300*time.Millisecond, oneAttemptBackOff) + resources := []string{"database", "deployment", "network", "volume"} + + type acquireResult struct { + resource string + lease *distlock.Lease + err error + } + resultCh := make(chan acquireResult, len(resources)) + start := make(chan struct{}) + acquireCtx, cancelAcquire := context.WithTimeout(context.Background(), 2*time.Second) + defer cancelAcquire() + for _, resource := range resources { + go func() { + <-start + lease, err := locker.Acquire(acquireCtx, resource) + resultCh <- acquireResult{resource: resource, lease: lease, err: err} + }() + } + close(start) + + results := make([]acquireResult, 0, len(resources)) + leases := make([]*distlock.Lease, 0, len(resources)) + for range resources { + result := requireReceive(t, resultCh, 2*time.Second, "concurrent acquisition did not finish") + results = append(results, result) + if result.lease != nil { + leases = append(leases, result.lease) + } + } + defer func() { + for _, lease := range leases { + releaseLease(t, lease) + } + }() + + for _, result := range results { + require.NoErrorf(t, result.err, "acquire %q", result.resource) + require.NotNilf(t, result.lease, "acquire %q", result.resource) + requireContextActive(t, result.lease.Context(), "independent lease was lost") + } +} + +func TestLockerExcludesCompetingLeaseUntilRelease(t *testing.T) { + cluster := newMemoryCluster(3) + firstLocker := newTestLocker(t, cluster, 300*time.Millisecond, retryBackOff) + secondLocker := newTestLocker(t, cluster, 300*time.Millisecond, retryBackOff) + + firstLease := acquireLease(t, firstLocker, "resource") + + type acquireResult struct { + lease *distlock.Lease + err error + } + resultCh := make(chan acquireResult, 1) + acquireCtx, cancelAcquire := context.WithTimeout(context.Background(), time.Second) + defer cancelAcquire() + go func() { + lease, acquireErr := secondLocker.Acquire(acquireCtx, "resource") + resultCh <- acquireResult{lease: lease, err: acquireErr} + }() + + requireNoReceive(t, resultCh, 50*time.Millisecond, "competing acquisition returned before release") + releaseLease(t, firstLease) + require.ErrorIs(t, context.Cause(firstLease.Context()), distlock.ErrLeaseReleased) + + result := requireReceive(t, resultCh, time.Second, "competing acquisition did not finish after release") + require.NoError(t, result.err) + require.NotNil(t, result.lease) + releaseLease(t, result.lease) +} + +func TestLockerContendingAcquireRespectsContext(t *testing.T) { + cluster := newMemoryCluster(3) + firstLocker := newTestLocker(t, cluster, 300*time.Millisecond, retryBackOff) + secondLocker := newTestLocker(t, cluster, 300*time.Millisecond, retryBackOff) + + firstLease := acquireLease(t, firstLocker, "resource") + t.Cleanup(func() { + if firstLease.Context().Err() != nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _ = firstLease.Release(ctx) + }) + + acquireCtx, cancelAcquire := context.WithTimeout(context.Background(), 75*time.Millisecond) + defer cancelAcquire() + competingLease, err := secondLocker.Acquire(acquireCtx, "resource") + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Nil(t, competingLease) + requireContextActive(t, firstLease.Context(), "holding lease was affected by a competing acquisition") + + releaseLease(t, firstLease) + secondLease := acquireLease(t, secondLocker, "resource") + releaseLease(t, secondLease) +} + +func TestLockerAutomaticallyRenewsLease(t *testing.T) { + const leaseDuration = 300 * time.Millisecond + cluster := newMemoryCluster(3) + locker := newTestLocker(t, cluster, leaseDuration, retryBackOff) + competingLocker := newTestLocker(t, cluster, leaseDuration, oneAttemptBackOff) + + lease := acquireLease(t, locker, "resource") + defer releaseLease(t, lease) + + time.Sleep(leaseDuration + 100*time.Millisecond) + requireContextActive(t, lease.Context(), "lease was lost instead of renewed") + + competingLease, err := competingLocker.Acquire(context.Background(), "resource") + require.Error(t, err) + require.Nil(t, competingLease) +} + +func TestLockerRenewsLeaseWithMinorityUnavailable(t *testing.T) { + const leaseDuration = 300 * time.Millisecond + cluster := newMemoryCluster(3) + locker := newTestLocker(t, cluster, leaseDuration, oneAttemptBackOff) + competingLocker := newTestLocker(t, cluster, leaseDuration, oneAttemptBackOff) + + lease := acquireLease(t, locker, "resource") + unavailableObserved := cluster.nodes[2].makeUnavailable() + originalValidityElapsed := time.NewTimer(leaseDuration + 100*time.Millisecond) + defer originalValidityElapsed.Stop() + defer func() { + cluster.nodes[2].setAvailable(true) + releaseLease(t, lease) + }() + + requireSignal(t, unavailableObserved, time.Second, "unavailable node did not receive renewal") + <-originalValidityElapsed.C + requireContextActive(t, lease.Context(), "lease was lost after a minority node became unavailable") + + competingLease, err := competingLocker.Acquire(context.Background(), "resource") + require.Error(t, err) + require.Nil(t, competingLease) +} + +func TestLockerLosesLeaseWhenNodesRestart(t *testing.T) { + const leaseDuration = 300 * time.Millisecond + cluster := newMemoryCluster(3) + locker := newTestLocker(t, cluster, leaseDuration, oneAttemptBackOff) + + lease := acquireLease(t, locker, "resource") + for _, node := range cluster.nodes { + node.restart() + } + + select { + case <-lease.Context().Done(): + require.ErrorIs(t, context.Cause(lease.Context()), distlock.ErrLeaseLost) + case <-time.After(time.Second): + require.FailNow(t, "lease was not lost after its node state disappeared") + } + + releaseLease(t, lease) + require.ErrorIs(t, context.Cause(lease.Context()), distlock.ErrLeaseLost) +} diff --git a/pkg/distlock/node.go b/pkg/distlock/node.go new file mode 100644 index 00000000..e8921198 --- /dev/null +++ b/pkg/distlock/node.go @@ -0,0 +1,29 @@ +package distlock + +import ( + "context" + "time" +) + +// Cluster provides point-in-time snapshots of the independent nodes that participate in distributed leases. +// Implementations must be safe for concurrent use. +type Cluster interface { + // Nodes returns each node counted toward quorum exactly once, including nodes that are temporarily unavailable. + // The Locker may keep and use the returned nodes throughout acquisition and until any acquired lease is released. + Nodes(ctx context.Context) ([]Node, error) +} + +// Node provides lease operations on a single cluster node. +// +// Each operation performs one attempt. A true result means the operation took effect. A false result with no error +// means the node responded but its lease state rejected the operation. If err is non-nil, the result is unknown and +// the boolean result must be ignored. Implementations must return promptly when ctx is cancelled, not modify token, +// and be safe for concurrent use. +type Node interface { + // Acquire creates a lease when resource does not have an unexpired lease. + Acquire(ctx context.Context, resource string, token []byte, ttl time.Duration) (bool, error) + // Renew extends an unexpired lease when its ownership token matches. + Renew(ctx context.Context, resource string, token []byte, ttl time.Duration) (bool, error) + // Release removes an unexpired lease when its ownership token matches. + Release(ctx context.Context, resource string, token []byte) (bool, error) +}