Idempotency of in-flight operations in Go

← Back to the section

When a service receives SIGTERM, it begins a graceful shutdown: it stops accepting new requests, waits for the current ones, and stops. Most operations manage to finish within the allotted 25–30 seconds. But if an operation did not fit — it can be retried only under one condition: the retry must not create a duplicate.

This is exactly what idempotency is — the property of an operation for which performing it once or several times is equally safe.

Why graceful shutdown alone is not enough

Imagine: a service processes an order and sends a request to charge money. At that moment SIGTERM arrives. http.Server.Shutdown in Go waits for the HTTP connections to finish, but it has a timeout — usually 25–30 seconds. If the charge did not finish in that time, the service stops forcibly.

A new instance of the service comes up and processes the same order again. If the charge request is not idempotent — the money will be charged twice.

Graceful shutdown gives time, but does not guarantee completeness. Idempotency is the insurance for the case when there was not enough time.

An outbound HTTP request: the Idempotency-Key

The most common case is a service calling an external payment provider or another service over HTTP. On a retry, the server must understand: "I have already processed this request".

For this, the Idempotency-Key header is used — a unique key that the client puts on the request. The server remembers the key and, on a retry, returns the previous result without performing the operation again.

The main rule: the key is generated once before any retry attempts, not on every attempt.

// internal/adapters/out/payment/client.go

type ChargeCommand struct {
    IdempotencyKey string
    OrderID        string
    CustomerID     string
    AmountKopecks  int64
}

func (c *Client) Charge(ctx context.Context, cmd ChargeCommand) error {
    body, _ := json.Marshal(map[string]any{
        "order_id":    cmd.OrderID,
        "customer_id": cmd.CustomerID,
        "amount":      cmd.AmountKopecks,
    })

    req, _ := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/charges", bytes.NewReader(body))
    req.Header.Set("Content-Type", "application/json")
    req.Header.Set("Idempotency-Key", cmd.IdempotencyKey) // one key for the whole operation

    resp, err := c.http.Do(req)
    if err != nil {
        return fmt.Errorf("charge request: %w", err)
    }
    defer resp.Body.Close()

    if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
        return fmt.Errorf("charge failed: status %d", resp.StatusCode)
    }
    return nil
}

The key is formed at the business-logic level — deterministically, from the data of the operation itself:

// internal/usecase/confirm_order.go

func (uc *ConfirmOrderUseCase) Execute(ctx context.Context, cmd ConfirmOrderCommand) error {
    order, err := uc.orders.FindByID(ctx, cmd.OrderID)
    if err != nil {
        return fmt.Errorf("find order: %w", err)
    }

    // Deterministic key: the same one for the same request
    idempotencyKey := order.ID + ":charge:" + cmd.RequestID

    return uc.payment.Charge(ctx, payment.ChargeCommand{
        IdempotencyKey: idempotencyKey,
        OrderID:        order.ID,
        CustomerID:     order.CustomerID,
        AmountKopecks:  order.TotalKopecks,
    })
}

A common mistake is to call uuid.New() on every attempt. Then every retry goes out with a new key, and the provider treats it as a separate operation.

Kafka consumer: deduplication via processed_event

kafka-go does not commit the offset automatically — the commit is done explicitly via CommitMessages. If the service received a message from Kafka, processed it, but did not manage to commit the offset before SIGTERM — the next consumer run will read the same message again.

The standard solution: a processed_event table, into which a record is written in the same transaction as the main action. If the message arrived again — the record already exists, and we skip the operation.

// internal/consumer/order_consumer.go

func (c *OrderConsumer) handle(ctx context.Context, msg kafka.Message) error {
    var event OrderConfirmedEvent
    if err := json.Unmarshal(msg.Value, &event); err != nil {
        return fmt.Errorf("unmarshal: %w", err)
    }

    tx, err := c.pool.Begin(ctx)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(ctx)

    q := c.queries.WithTx(tx)

    // Record the fact of processing — in the same transaction as the action itself
    inserted, err := q.InsertProcessedEvent(ctx, db.InsertProcessedEventParams{
        EventID:    event.ID,
        ConsumerID: "billing-service",
    })
    if err != nil {
        return fmt.Errorf("insert processed event: %w", err)
    }
    if !inserted {
        return nil // already processed — skip
    }

    if err := c.billing.Charge(ctx, payment.ChargeCommand{
        IdempotencyKey: event.ID + ":billing",
        OrderID:        event.OrderID,
        CustomerID:     event.CustomerID,
        AmountKopecks:  event.TotalKopecks,
    }); err != nil {
        return fmt.Errorf("charge order %s: %w", event.OrderID, err)
    }

    return tx.Commit(ctx)
}

The SQL query for deduplication needs attention: you cannot use ON CONFLICT DO NOTHING, because then RETURNING on a conflict will return 0 rows, which leads to a pgx.ErrNoRows error. The correct variant is DO UPDATE, which always returns a row:

-- name: InsertProcessedEvent :one
INSERT INTO processed_event (event_id, consumer_id, processed_at)
VALUES (@event_id, @consumer_id, now())
ON CONFLICT (event_id, consumer_id) DO UPDATE
    SET processed_at = EXCLUDED.processed_at
RETURNING (xmax = 0) AS inserted;

On the first insert xmax = 0 and inserted = true is returned. On a retry the row is updated, xmax becomes nonzero, and inserted = false is returned.

Outbox-relay: two-phase publishing

The outbox is a table of events that need to be sent to Kafka. A relay goroutine periodically reads from it and publishes. The problem: if a message is sent to Kafka but the status in the table is not updated before SIGTERM — the relay will send it again.

There are two approaches.

If all downstream consumers implement processed_event — the relay can publish duplicates, and the consumer will filter them out. This is simpler:

// internal/scheduler/outbox_relay.go

func (r *OutboxRelay) processOneBatch(ctx context.Context) error {
    // context.Background() — so that SIGTERM does not interrupt the commit of a transaction that has started.
    // New batches are not started thanks to the ctx.Done() check in Run.
    bgCtx := context.Background()

    tx, err := r.pool.Begin(bgCtx)
    if err != nil {
        return fmt.Errorf("begin tx: %w", err)
    }
    defer tx.Rollback(bgCtx)

    q := r.queries.WithTx(tx)
    events, err := q.LockPendingOutboxBatch(bgCtx, r.batchSize) // FOR UPDATE SKIP LOCKED
    if err != nil {
        return fmt.Errorf("lock outbox batch: %w", err)
    }

    for _, e := range events {
        if err := r.writer.WriteMessages(ctx, kafka.Message{
            Key:   []byte(e.AggregateID),
            Value: e.Payload,
            Headers: []kafka.Header{
                {Key: "event-id", Value: []byte(e.ID)},
            },
        }); err != nil {
            return fmt.Errorf("write outbox event %s: %w", e.ID, err)
        }

        if err := q.MarkOutboxDispatched(bgCtx, e.ID); err != nil {
            return fmt.Errorf("mark dispatched %s: %w", e.ID, err)
        }
    }

    return tx.Commit(bgCtx)
}

If the downstream consumer is not under control — you need explicit two-phase publishing through the statuses PENDING → PUBLISHING → PUBLISHED:

-- Move to PUBLISHING atomically (FOR UPDATE SKIP LOCKED)
-- name: LockAndMarkPublishing :many
UPDATE outbox_event
SET status = 'PUBLISHING', locked_at = now()
WHERE id IN (
    SELECT id FROM outbox_event
    WHERE status = 'PENDING'
    ORDER BY created_at
    LIMIT @batch_size
    FOR UPDATE SKIP LOCKED
)
RETURNING *;

-- Move to PUBLISHED after a successful send
-- name: MarkPublished :exec
UPDATE outbox_event SET status = 'PUBLISHED', published_at = now() WHERE id = @id;

If SIGTERM arrived between the send and MarkPublished, the rows are stuck in PUBLISHING. A cleanup goroutine returns them to PENDING after a configurable TTL:

func (c *OutboxCleanup) Run(ctx context.Context, wg *sync.WaitGroup) {
    defer wg.Done()
    ticker := time.NewTicker(c.interval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            if err := c.queries.ResetStuckPublishing(ctx, db.ResetStuckPublishingParams{
                StuckAfter: pgtype.Interval{Microseconds: int64(c.stuckAfter / time.Microsecond), Valid: true},
            }); err != nil {
                slog.WarnContext(ctx, "outbox cleanup failed", "error", err)
            }
        }
    }
}

-- name: ResetStuckPublishing :exec
UPDATE outbox_event
SET status = 'PENDING', locked_at = NULL
WHERE status = 'PUBLISHING'
  AND locked_at < now() - @stuck_after::interval;

context.Background() in the relay — why

The relay goroutine gets a cancellable context that is cancelled on SIGTERM. But if you pass that same context into a pgx transaction — SIGTERM will cancel the transaction right at the moment of commit, and the batch will roll back.

The solution: use context.Background() for the transaction inside the batch, and do the cancellation check between iterations:

func (r *OutboxRelay) Run(ctx context.Context, wg *sync.WaitGroup) {
    defer wg.Done()
    ticker := time.NewTicker(r.interval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return // do not start a new batch
        case <-ticker.C:
        }

        batchCtx, cancel := context.WithTimeout(context.Background(), r.batchTimeout)
        if err := r.processOneBatch(batchCtx); err != nil {
            slog.WarnContext(ctx, "outbox relay batch", "error", err)
        }
        cancel()
    }
}

ctx.Done() is the signal to stop the loop. context.Background() for the batch is so that a transaction that has already started can commit.

In short

Graceful shutdown gives time to finish operations, but does not guarantee it. Operations must be safe to retry — this is exactly what idempotency is.
For an outbound HTTP POST: set the Idempotency-Key header in the adapter. The key is formed deterministically once per business operation, and is not regenerated on every attempt.
For a Kafka consumer: record the fact of processing in a processed_event table in the same pgx transaction as the main action. SQL — ON CONFLICT DO UPDATE ... RETURNING (xmax = 0) AS inserted (not DO NOTHING — on a conflict DO NOTHING does not return a row).
For the outbox-relay: if the downstream is not under control — a two-phase status PENDING → PUBLISHING → PUBLISHED plus a cleanup goroutine for stuck records.
In the relay goroutine, use context.Background() for the pgx transaction inside the batch, and do the cancellation check between iterations via select { case <-ctx.Done(): return }.