DEV Community

RaffertyBarrett4726
RaffertyBarrett4726

Posted on

Implementing Distributed Data Cleanup Across Multiple Worker Nodes (Lease Fencing)

#sre #distributedsystems #scheduling #dataintegrity

Nightly payment reconciliation should be treated as a recovery problem, not as a timer problem. For a customer-support system, let worker nodes compete for small, explicitly leased cleanup units, make every mutation idempotent, and record enough state to retry safely. The latency-versus-cost choice is then visible: add workers when the reconciliation window is at risk, but keep coordination and cleanup contracts stable.

TL;DR: schedule one run, partition the payment-provider scope into work items, claim each item with a lease and a fencing token, and make cleanup conditional on that token. A worker that loses its lease may finish its process, but it must not commit.

I have been paged for both shapes: a missed job and a duplicate delivery. Both happen when a distributed scheduler is treated as if it has one clock and one executor. Two worker nodes can be healthy and still disagree about ownership.

The incident lesson: a timer is not ownership

A nightly run can be started once and still execute more than once. A process can claim work and pause before its network call. Another node can take over after the lease expires while the first node is still alive. A queue visibility timeout has the same operational implication: hiding a message temporarily does not prove that its consumer completed the work. The consumer must acknowledge it after success, and a timeout can make it visible again.

The invariant I want is narrower than “exactly once,” because exactly-once execution is not a useful promise at this boundary:

Every cleanup effect is safe to repeat, and a worker without current ownership cannot commit a stale result.

The scheduler creates a run record with a deterministic business date. A dispatcher expands it into items such as account_id + provider_cursor + date. Workers claim items with a conditional write. Each item has a lease expiry, an attempt count, and a fencing token that increases on every successful claim.

I initially treated the lease as a lock. That was the wrong mental model. A lock says who may enter; a lease says who may enter until a deadline, while a delayed process may still hold an old copy. Fencing makes that old copy harmless.

How can I implement distributed data cleanup across multiple nodes?

Keep an item small enough that one retry has a bounded effect. One item can represent one support account and one reconciliation cursor, rather than an entire provider export. The durable record needs a stable identity, state, owner, lease expiry, fencing token, and last error. The storage technology is less important than conditional transition semantics.

A claim must atomically change the owner and increment the token. A plain read followed by a plain write leaves a race between nodes. The cleanup mutation should include the item identity and token in its predicate; if no record matches, the worker has lost ownership and must stop issuing writes.

Here is the shape of that boundary in Go. The repository exposes conditional operations instead of pretending that a process-local mutex coordinates a worker fleet.

package cleanup

import (
    "context"
    "errors"
    "time"
)

var ErrLeaseLost = errors.New("cleanup lease lost")

type Store interface {
    Claim(context.Context, string, string, time.Time, time.Duration) (Lease, error)
    ApplyIfOwner(context.Context, string, int64, time.Time, Effect) error
    MarkDoneIfOwner(context.Context, string, int64, time.Time) error
}

type Lease struct {
    ItemID string
    Token  int64
}

type Effect struct {
    AccountID string
    Cursor    string
    RunDate   string
}

func Process(ctx context.Context, store Store, workerID, itemID string) error {
    lease, err := store.Claim(ctx, itemID, workerID, time.Now().UTC(), 10*time.Minute)
    if err != nil {
        return err
    }
    effect, err := loadEffect(ctx, itemID)
    if err != nil {
        return err
    }
    if err := store.ApplyIfOwner(ctx, lease.ItemID, lease.Token, time.Now().UTC(), effect); err != nil {
        if errors.Is(err, ErrLeaseLost) {
            return nil
        }
        return err
    }
    return store.MarkDoneIfOwner(ctx, lease.ItemID, lease.Token, time.Now().UTC())
}

func loadEffect(context.Context, string) (Effect, error) { return Effect{}, nil }
Enter fullscreen mode Exit fullscreen mode

The provider call and durable mutation belong behind separate interfaces, and their failure behavior must be tested against the selected store. In production, ApplyIfOwner should carry an idempotency key derived from the cleanup item. Reuse it across retries when the external provider supports such a key. If it does not, make the local effect record authoritative and reconcile an ambiguous response before issuing another mutation.

How do you choose latency versus cost?

Start with the reconciliation deadline and the size of one work item. If one worker can finish with retry headroom, adding a fleet buys little. If the deadline is threatened, parallelize independent accounts, cap concurrency at the provider boundary, and measure the oldest pending item instead of only measuring average duration. More workers reduce elapsed time until the provider, storage, or retry queue becomes the limiting resource.

The cost is not only compute. More concurrent attempts create more provider requests, more coordination writes, and more rate-limit events. A cheap schedule that misses the support team's morning window is expensive operationally. A large always-on fleet is wasteful when the nightly set is small.

My decision rule is simple:

Signal Action Reason
Backlog age is healthy Keep the normal worker cap Extra concurrency adds coordination cost
Backlog age approaches the deadline Add bounded burst capacity Trade compute for completion headroom
Provider throttling rises Reduce concurrency and retry with delay More workers would increase latency
Stale claims rise Inspect lease duration and pauses Capacity cannot repair unsafe ownership

Do not silently declare success because the dispatcher ran. Finish with explicit partial or failed state when the deadline passes.

When does this lease pattern not apply?

The limitation is important: do not use a lease as a substitute for a transaction when the cleanup effect and its completion record must change atomically in the same database. Use the database transaction there, then make the external side idempotent or reconcile its uncertain result.

This pattern also does not make a provider call cancellable after a worker loses its lease. It is a poor fit when the cleanup effect and its status require one atomic transaction that the external boundary cannot support; in that case, choose a transactional local outbox or a provider-specific reconciliation process. Cancellation can reduce wasted work, but fencing is still required because cancellation can arrive late or fail. A lease should exceed the normal critical section with room for storage and provider latency; renewal is appropriate for long items, provided renewal is conditional on the current token.

Test the ugly transitions directly: a process pauses after claiming, the lease expires, another node claims the item, the first process resumes, the provider times out after accepting the request, and the final status write fails. The expected result is not that only one attempt ran. The expected result is that stale attempts cannot create a second cleanup effect and that the item remains recoverable.

References

Sources

Top comments (0)