Swig 🍺

Advanced Concepts

Deep dive into Swig's advanced features and architecture

Advanced Concepts

Transactional Integrity

In distributed systems, especially job queues, transactional integrity is crucial. Swig offers three levels of transaction control:

Use your existing database transaction for maximum control:

tx, _ := pool.Begin(ctx)
defer tx.Rollback(ctx)
 
// Create user
userID := createUser(tx)
 
// Enqueue welcome email in same transaction
err := swigClient.AddJobWithTx(ctx, tx, &EmailWorker{
    To: email,
    Subject: "Welcome!",
})
if err != nil {
    return err
}
return tx.Commit(ctx)

2. Use Swig's Transaction Helper

Let Swig manage the transaction for you:

err := swigClient.driver.WithTx(ctx, func(tx Transaction) error {
    // Create user
    if err := createUser(tx); err != nil {
        return err // Triggers rollback
    }
    
    // Add job (auto rollback if this fails)
    return tx.Exec(ctx, insertJobSQL, ...)
})

3. No Transaction (Simple)

For simple, non-transactional job enqueueing:

err := swigClient.AddJob(ctx, &EmailWorker{
    To: email,
    Subject: "Welcome!",
})

Benefits of Transactional Job Processing

  • No Lost Jobs: Jobs are either fully committed or not at all
  • Atomic Claiming: Workers claim jobs with PostgreSQL's SKIP LOCKED row locks
  • Data Consistency: Jobs can be part of your application's transactions

Architecture

Swig uses PostgreSQL's advanced features for efficient job distribution:

Worker Factories

Workers are registered once by job name, but Swig creates a fresh worker instance for each claimed job before unmarshaling the job payload. This keeps concurrent jobs from mutating the same registered worker value.

SKIP LOCKED for Job Distribution

Swig uses PostgreSQL's FOR UPDATE SKIP LOCKED feature to ensure:

  • No duplicate job processing
  • Fair job distribution across workers
  • High availability
  • Transactional integrity

The row lock plays a similar coordination role to a mutex, but at the database level. A Go mutex protects memory inside one process. PostgreSQL row locks protect shared job rows across processes and machines.

Dedicated LISTEN/NOTIFY Connections

PostgreSQL LISTEN is session-scoped. A listener must wait for notifications on the same connection that executed LISTEN.

Swig now creates a dedicated listener for each worker. The listener stays open for the worker lifetime, receives swig_jobs notifications, and is closed during shutdown. This avoids subtle connection-pool bugs where LISTEN happens on one pooled connection but notification waiting happens on another.

Leader Election

Built-in leader election using PostgreSQL advisory locks ensures:

  • Only one Swig instance performs leader duties at a time
  • Automatic failover if the leader fails
  • No external coordination needed

Advisory locks are also session-scoped, so Swig acquires the leader lock on a dedicated connection and keeps that connection open while leader duties run. On shutdown, Swig releases the advisory lock from the same session that acquired it.

Instances that do not become leader keep retrying leadership in the background. If the active leader connection dies, PostgreSQL releases the advisory lock and another Swig instance can acquire leadership on a later retry.

Batch Processing

Swig supports efficient batch insertion of multiple jobs:

Basic Batch Insertion

batchJobs := []drivers.BatchJob{
    {
        Worker: &EmailWorker{
            To: "batch1@example.com",
            Subject: "Batch Welcome",
            Body: "Welcome to our platform!",
        },
        Opts: drivers.JobOptions{
            Queue: "default",
            Priority: 1,
            RunAt: time.Now(),
        },
    },
    {
        Worker: &EmailWorker{
            To: "batch2@example.com",
            Subject: "Batch Welcome",
            Body: "Welcome to our platform!",
        },
        Opts: drivers.JobOptions{
            Queue: "default",
            Priority: 1,
            RunAt: time.Now(),
        },
    },
}
 
err := swigClient.AddJobs(ctx, batchJobs)
if err != nil {
    log.Printf("Failed to add batch jobs: %v", err)
} else {
    log.Println("Batch jobs added successfully")
}

Transactional Batch Insertion

tx, err := db.Begin()
if err != nil {
    log.Printf("Failed to begin transaction: %v", err)
    return
}
defer tx.Rollback()
 
txBatchJobs := []drivers.BatchJob{
    {
        Worker: &EmailWorker{
            To: "tx1@example.com",
            Subject: "Transactional Welcome",
            Body: "Welcome to our platform!",
        },
        Opts: drivers.JobOptions{
            Queue: "default",
            Priority: 1,
            RunAt: time.Now(),
        },
    },
    {
        Worker: &EmailWorker{
            To: "tx2@example.com",
            Subject: "Transactional Welcome",
            Body: "Welcome to our platform!",
        },
        Opts: drivers.JobOptions{
            Queue: "default",
            Priority: 1,
            RunAt: time.Now(),
        },
    },
}
 
log.Println("Adding transactional batch jobs...")
if err := swigClient.AddJobsWithTx(ctx, tx, txBatchJobs); err != nil {
    log.Printf("Failed to add transactional batch jobs: %v", err)
    return
}
 
if err := tx.Commit(); err != nil {
    log.Printf("Failed to commit transaction: %v", err)
    return
}
log.Println("Transactional batch jobs added successfully")

Performance Considerations

Batch Processing

  • Batch insertion is significantly faster than individual inserts
  • Optimal batch size depends on your database configuration
  • Monitor memory usage for large batches

Driver Selection

Swig supports two PostgreSQL drivers with different characteristics:

  1. pgx Driver (Recommended)

    • Better performance
    • Native LISTEN/NOTIFY support
    • Real-time job notifications
    • Optimized for batch operations

  2. database/sql Driver

    • Standard Go database interface
    • Requires github.com/lib/pq for LISTEN/NOTIFY
    • Simpler implementation
    • Good for most use cases

Cleanup and Testing

Swig provides methods for both graceful shutdown and complete cleanup:

// Graceful shutdown: cancel workers, wait for them, and release leader resources
err := swigClient.Stop(ctx)
 
// Complete cleanup: Drop all Swig tables
err := swigClient.Close(ctx)

The Close method is particularly useful in:

  • Testing environments
  • Development scenarios
  • CI/CD pipelines

Next Steps

API Reference

Detailed API documentation for Swig

On this page