Do Not Enqueue Before Transaction Commit

Introduction

Performing both database writes and message queue enqueues within the same transaction (hereafter TX) can lead to race conditions. This article analyzes an issue that actually occurred in code that enqueued inside a TX, and explains the fix pattern of moving the enqueue operation to after the transaction commits.

Target audience: Backend engineers implementing services with Go + GORM + a message queue.

Scope: This article covers race conditions caused by enqueuing inside a TX, the fix pattern of post-commit enqueue, and the resulting changes in error handling. Automatic recovery mechanisms for enqueue failures, such as the Outbox pattern, are out of scope.

Terminology

The following terms are used throughout this article.

TX Transaction. A unit that executes a series of database operations atomically.

enqueue The operation of publishing a message to a message queue.

dequeue The operation of consuming a message from a message queue.

consumer A service that dequeues and processes messages from the queue.

commit The operation of persisting a TX’s changes to the database.

rollback The operation of discarding a TX’s changes.

race condition A state where the outcome depends on the timing of concurrent operations.

PENDING record An unprocessed record left in the database when an enqueue fails.

message queue A system for asynchronous message delivery between services. This article assumes a queue that does not participate in DB transactions.

What Happens with Pre-Commit Enqueue

When an enqueue is performed inside a TX, the consumer may receive the message before the TX has been committed. Since the target record is not yet committed and therefore not visible in the database, the consumer cannot process the message correctly.

Below are two concrete reproduction scenarios.

Scenario 1: Single Record INSERT + Enqueue

This is a case where a record is inserted into the publish_requests table and the consumer is asked to process that record.

Timeline →

[Producer]                                [Consumer]
  │                                         │
  ├─ BEGIN TX                               │
  ├─ INSERT publish_requests (uncommitted)  │
  ├─ enqueue(message) ──────────────────→   dequeue(message)
  │                                         ├─ GetByID → record not visible
  │                                         └─ message discarded
  └─ COMMIT

1. A record is inserted into publish_requests inside the TX (not yet committed)
2. An enqueue is performed inside the TX (the message is delivered immediately)
3. The consumer dequeues the message and attempts to retrieve the record via GetByID
4. Because the TX has not been committed, the record is not visible in the database
5. The consumer determines that the record does not exist and discards the message
6. The TX commits afterward, but the message has already been discarded

Scenario 2: Bulk INSERT + Enqueue

This is a case where multiple records are bulk-inserted into the outbox table and the consumer is asked to process the PENDING records in bulk.

Timeline →

[Producer]                                [Consumer]
  │                                         │
  ├─ BEGIN TX                               │
  ├─ Bulk INSERT outbox (uncommitted)       │
  ├─ enqueue(message) ──────────────────→   dequeue(message)
  │                                         ├─ FindPending → empty array
  │                                         └─ message discarded
  └─ COMMIT

1. Multiple records are bulk-inserted into the outbox table inside the TX (not yet committed)
2. An enqueue is performed inside the TX (the message is delivered immediately)
3. The consumer dequeues the message and searches for PENDING records via FindPending
4. Because the TX has not been committed, the bulk-inserted records are not visible, and the result is an empty array
5. The consumer determines there is nothing to process and discards the message

In both scenarios, the root cause is the same. By the time the enqueued message reaches the consumer, the TX has not yet been committed, so the records are not visible to the consumer.

Code Structure Before the Fix

In the original code, DB operations and the enqueue were executed within the same TX. Below is a code example corresponding to Scenario 1.

func (s *Service) ExecutePublishTransaction(ctx context.Context, jobID uint) error {
    return s.db.Transaction(func(tx *gorm.DB) error {
        return s.executePublish(ctx, tx, jobID)
    })
}

func (s *Service) executePublish(ctx context.Context, tx *gorm.DB, jobID uint) error {
    // 1. INSERT into publish_requests
    req := &PublishRequest{
        JobID: jobID,
        Status:  "PENDING",
    }
    if err := tx.Create(req).Error; err != nil {
        return err
    }
    // 2. UPDATE jobs table status
    if err := tx.Model(&Job{}).
        Where("id = ?", jobID).
        Update("status", "PROCESSING").Error; err != nil {
        return err
    }
    // 3. Build queue message and enqueue (still inside the TX)
    body, err := json.Marshal(QueueMessage{
        Type: "PROCESS",
        ID:   req.ID,
    })
    if err != nil {
        return err
    }
    if err := s.queueRepo.Enqueue(ctx, body); err != nil {
        return err // enqueue failure → entire TX rolls back
    }
    return nil
}

This code had the following annotation:

NOTE: Publish runs inside a TX. Orphaned messages from TX failures are safely skipped by idempotency checks on the consumer side.

Enqueuing inside the TX was an intentional design choice. The assumption was that messages left in the queue after a TX failure would be safely handled by the consumer’s idempotency checks. However, the race condition described above was not accounted for.

Fix Pattern: Post-Commit Enqueue

The fix is to restrict the TX-internal function to DB operations only and move the enqueue to after the TX commit.

Step 1: Restrict the TX-Internal Function to DB Operations Only

The function was renamed from executePublish to createPublishRequestInTx, and the queue repository was removed from its parameters. This function now performs only DB operations.

func (s *Service) createPublishRequestInTx(ctx context.Context, tx *gorm.DB, jobID uint) (*PublishRequest, error) {
    // 1. INSERT into publish_requests
    req := &PublishRequest{
        JobID: jobID,
        Status:  "PENDING",
    }
    if err := tx.Create(req).Error; err != nil {
        return nil, err
    }

    // 2. UPDATE jobs table status
    if err := tx.Model(&Job{}).
        Where("id = ?", jobID).
        Update("status", "PROCESSING").Error; err != nil {
        return nil, err
    }
    return req, nil // no enqueue here
}

Two differences from the original executePublish:

• The queue repository was removed from the parameters (restricting the function to DB operations only)
• The created PublishRequest is returned as a return value (to be used for the post-commit enqueue)

Step 2: Enqueue After the Commit

The calling method ExecutePublishTransaction now performs the enqueue after the TX commit.

func (s *Service) ExecutePublishTransaction(ctx context.Context, jobID uint) error {
    var req *PublishRequest

    // Only DB operations inside the TX
    err := s.db.Transaction(func(tx *gorm.DB) error {
        var txErr error
        req, txErr = s.createPublishRequestInTx(ctx, tx, jobID)
        return txErr
    })
    if err != nil {
        return err // TX failed → DB operations already rolled back
    }
    // Enqueue after COMMIT.
    // Enqueuing before COMMIT would leave the record uncommitted,
    // causing a race condition where the consumer processes the
    // message and cannot see the record, discarding it.
    // Therefore the enqueue is performed outside the TX.
    body, err := json.Marshal(QueueMessage{
        Type: "PROCESS",
        ID:   req.ID,
    })
    if err != nil {
        s.logger.WithField("publish_request_id", req.ID).
            WithError(err).
            Error("failed to marshal queue message")
        return nil
    }
    if err := s.queueRepo.Enqueue(ctx, body); err != nil {
        // On enqueue failure, DB operations are already committed
        // and cannot be rolled back.
        // Log the error and continue.
        s.logger.WithField("publish_request_id", req.ID).
            WithError(err).
            Error("failed to enqueue after commit")
        return nil
    }
    return nil
}

The annotation was updated to:

DB operations are executed atomically inside the TX; the enqueue is performed after COMMIT.

Changes in Error Handling

Moving the enqueue from inside the TX to outside it changes the behavior on enqueue failure.

Before (enqueue inside TX)After (enqueue outside TX)enqueue succeedsThe entire TX is committedTX is already committed; no changeenqueue failsError returned → entire TX rolls back → DB operations are undoneOnly a log entry → DB operations remain committed

Before the fix, an enqueue failure caused the entire TX to roll back, undoing all DB operations including the INSERT and UPDATE. Data consistency was preserved, but the race condition risk existed.

After the fix, even if the enqueue fails, the DB operations are already committed and cannot be rolled back. A record with PENDING status remains in the publish_requests table. Because this record was not enqueued, the consumer does not process it.

When this situation occurs, manual recovery is still possible using the publish_request_id logged in the error output. Specifically, the PENDING record can be identified and manually enqueued.

Automated recovery mechanisms are a topic for future work.

Test Changes

Because the behavior on enqueue failure changed, the test case assertions also need to be updated.

Tests Before the Fix

{
    name: "returns error on enqueue failure",
    setupMock: func(m *MockQueueRepo) {
        m.EXPECT().
            Enqueue(gomock.Any(), gomock.Any()).
            Return(errors.New("queue unavailable"))
    },
    wantErr: true,
},
{
    name: "TX rolls back on enqueue failure and no DB operations remain",
    setupMock: func(m *MockQueueRepo) {
        m.EXPECT().
            Enqueue(gomock.Any(), gomock.Any()).
            Return(errors.New("queue unavailable"))
    },
    assertDB: func(t *testing.T, db *gorm.DB) {
        var count int64
        db.Model(&PublishRequest{}).Count(&count)
        assert.Equal(t, int64(0), count) // rolled back → 0 records
    },
    wantErr: true,
},

Tests After the Fix

{
    name: "does not return error on enqueue failure (log only)",
    setupMock: func(m *MockQueueRepo) {
        m.EXPECT().
            Enqueue(gomock.Any(), gomock.Any()).
            Return(errors.New("queue unavailable"))
    },
    wantErr: false, // assert.Error → assert.NoError
},
{
    name: "does not return error on enqueue failure; DB operations remain because TX is already committed",
    setupMock: func(m *MockQueueRepo) {
        m.EXPECT().
            Enqueue(gomock.Any(), gomock.Any()).
            Return(errors.New("queue unavailable"))
    },
    assertDB: func(t *testing.T, db *gorm.DB) {
        var count int64
        db.Model(&PublishRequest{}).Count(&count)
        assert.Equal(t, int64(1), count) // already committed → 1 record remains
    },
    wantErr: false,
},

Two changes:

1. Assertion change: assert.Error to assert.NoError. The function no longer returns an error on enqueue failure.
2. DB verification change: Record count from 0 (rolled back) to 1 (already committed). DB operations are no longer rolled back.

Summary

Enqueuing inside a TX introduces a race condition where the record is still uncommitted when the consumer receives the message. The record is not visible to the consumer, so the message is discarded.

To fix this issue, the following changes were made:

• Restrict the TX-internal function to DB operations only: Separate the enqueue logic and update the function name and parameters
• Move the enqueue to after the commit: Execute the enqueue only after the TX has successfully committed
• On enqueue failure, log and continue: DB operations are already committed and cannot be rolled back

This fix eliminates the race condition, but introduces a trade-off: there is no automatic recovery mechanism for enqueue failures. If a PENDING record remains, manual recovery is still possible using the ID from the logs. An automated recovery mechanism (such as Outbox table management) is being considered separately as future work.