Implementing a PostgreSQL-backed Commit Log

Introduction to the Lesson

Today, we will leverage the power of the abstraction we created. We are going to build a second, entirely different implementation of the exact same CommitLog interface, but this time backed by PostgreSQL.

Your Goal for Today: Your learning outcome is to implement the storage interface using a PostgreSQL-backed commit log, reusing your database skills.

This exercise will powerfully demonstrate the key benefit of interface-driven design: the ability to swap complex subsystems (like a storage engine) without changing the application logic that depends on them. We'll explore the trade-offs of using a relational database for this task and see how it can simplify concerns like durability and replication.

By the end of this lesson, you will have a PostgresLog that is a perfect, interchangeable replacement for the file-based Log we built previously.

1. The Repository Pattern in Action

What we are doing is a classic application of the Repository Pattern. This pattern mediates between the domain (our broker's business logic) and data mapping layers (the storage implementations).

The CommitLog interface acts as the contract. Our application code interacts only with this interface, remaining completely unaware of whether data is being written to a local file or a remote PostgreSQL server.

The main purpose is to decouple business logic from data access details. This makes the system more modular, easier to test, and flexible enough to evolve.

2. Designing the Commit Log Schema in PostgreSQL

To store our commit log in PostgreSQL, we need a table. How should we structure it to represent an append-only, partitioned log?

A single table can hold the messages for all partitions. We need columns to identify the partition, the message's offset within that partition, and the message data itself.

Here is a simple but effective schema. Let's call the table messages.

CREATE TABLE messages (
    partition_id TEXT NOT NULL,
    offset BIGSERIAL,
    data BYTEA NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (partition_id, offset)
);

Let's break this down:

• partition_id: A TEXT field to identify the topic partition this message belongs to (e.g., "orders_0", "orders_1").
• offset: A BIGSERIAL is an auto-incrementing 8-byte integer. This is perfect for our use case. When we insert a row for a given partition, PostgreSQL will automatically assign the next sequential offset for that table, which we can then return. We'll discuss a nuance with this shortly. The BIGSERIAL type gives us an auto-incrementing offset.
• data: The BYTEA type is for binary data, which is exactly what we need to store our marshaled protobuf Record.
• PRIMARY KEY (partition_id, offset): This composite primary key ensures that each offset is unique within its partition, and it creates an index that will make reads by partition and offset very fast.

A Note on BIGSERIAL and Partition-Specific Offsets:
The BIGSERIAL type increments across the entire table. However, our CommitLog interface expects offsets to be sequential within a single partition. For our simplified broker, we can either:

1. Manage a sequence per partition inside Postgres.
2. Have a separate table for each partition.
3. For simplicity in this lesson, we will build a simplified version where one table holds messages and the offset is managed manually in the application layer to ensure it's sequential per partition. Let's refine the schema for this approach.

Revised Schema:

CREATE TABLE messages (
    partition_id TEXT NOT NULL,
    "offset" BIGINT NOT NULL,
    data BYTEA NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (partition_id, "offset")
);

We've changed offset from BIGSERIAL to BIGINT. Our application will now be responsible for providing the correct offset on INSERT. This aligns more closely with the CommitLog abstraction, where the log implementation controls the offset sequence.

3. Implementing the PostgresLog

Let's create the Go implementation. We'll create a new file, internal/log/postgres_log.go. This struct will hold a database connection pool and the ID of the partition it manages.

Since you've already worked with pgx in a previous module, we'll use pgxpool for managing connections.

// internal/log/postgres_log.go
package log

import (
	"context"
	"fmt"

	"github.com/jackc/pgx/v5/pgxpool"
	"google.golang.org/protobuf/proto"
	
	api "github.com/your-username/your-project/api/v1" // Path to your generated protobuf code
)

// Ensure PostgresLog satisfies the CommitLog interface.
var _ CommitLog = (*PostgresLog)(nil)

type PostgresLog struct {
	pool        *pgxpool.Pool
	partitionID string
}

func NewPostgresLog(pool *pgxpool.Pool, partitionID string) (*PostgresLog, error) {
	log := &PostgresLog{
		pool:        pool,
		partitionID: partitionID,
	}
    // In a real system, you might ensure the partition exists or create it here.
	return log, nil
}

Implementing Append

The Append method needs to get the current highest offset for the partition, increment it, and then insert the new record with that offset. We can do this safely within a single transaction.

// Append adds a record to the messages table for the given partition.
func (l *PostgresLog) Append(record *api.Record) (uint64, error) {
	ctx := context.Background()
	var newOffset uint64

	// Start a transaction
	tx, err := l.pool.Begin(ctx)
	if err != nil {
		return 0, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback(ctx) // Rollback is a no-op if tx has been committed

	// 1. Get the current highest offset for this partition, locking the row to prevent race conditions.
	highestOffsetQuery := `
		SELECT "offset" FROM messages 
		WHERE partition_id = $1 
		ORDER BY "offset" DESC 
		LIMIT 1 FOR UPDATE;`
	
	var lastOffset int64 // Use signed int for scanning, as pgx might return it this way
	err = tx.QueryRow(ctx, highestOffsetQuery, l.partitionID).Scan(&lastOffset)
	
    // If no rows are found, this is the first record, so the offset is 0.
    // pgx.ErrNoRows is the specific error to check.
	if err != nil && err.Error() == "no rows in result set" {
		newOffset = 0
	} else if err != nil {
		return 0, fmt.Errorf("failed to query highest offset: %w", err)
	} else {
		newOffset = uint64(lastOffset) + 1
	}

	// 2. Marshal the record and insert the new row.
	record.Offset = newOffset
	data, err := proto.Marshal(record)
	if err != nil {
		return 0, fmt.Errorf("failed to marshal record: %w", err)
	}

	insertQuery := `
		INSERT INTO messages (partition_id, "offset", data) 
		VALUES ($1, $2, $3);`
	
	_, err = tx.Exec(ctx, insertQuery, l.partitionID, newOffset, data)
	if err != nil {
		return 0, fmt.Errorf("failed to insert record: %w", err)
	}

	// 3. Commit the transaction.
	if err = tx.Commit(ctx); err != nil {
		return 0, fmt.Errorf("failed to commit transaction: %w", err)
	}

	return newOffset, nil
}

Here, using SELECT ... FOR UPDATE is critical. It places a lock on the row(s) it finds, ensuring that no other concurrent transaction can read or modify the latest offset until our transaction is complete. This prevents a classic race condition where two Append calls could try to insert a record with the same offset.

Implementing Read

The Read method is more straightforward. It's a simple SELECT query to fetch the data for a specific partition and offset.

// Read retrieves a record from the messages table by its offset.
func (l *PostgresLog) Read(offset uint64) (*api.Record, error) {
	ctx := context.Background()
	query := `SELECT data FROM messages WHERE partition_id = $1 AND "offset" = $2;`

	var data []byte
	err := l.pool.QueryRow(ctx, query, l.partitionID, offset).Scan(&data)
	if err != nil {
        // Handle not found error specifically
		if err.Error() == "no rows in result set" {
			return nil, fmt.Errorf("offset out of range: %d", offset)
		}
		return nil, fmt.Errorf("failed to read record: %w", err)
	}

	record := &api.Record{}
	if err := proto.Unmarshal(data, record); err != nil {
		return nil, fmt.Errorf("failed to unmarshal record: %w", err)
	}

	return record, nil
}

Implementing Close

For this implementation, the connection pool (*pgxpool.Pool) is passed in from outside. The lifecycle of the pool should be managed by the main application, not by an individual log instance. Therefore, the Close method for PostgresLog can be a no-op.

// Close is a no-op for PostgresLog as the connection pool is managed externally.
func (l *PostgresLog) Close() error {
	return nil
}

4. The Log of Logs: Durability via PostgreSQL

When we built the file-based log, we had to consider low-level details like OS page caches and forcing writes to disk with fsync to ensure durability.

By using PostgreSQL, we delegate this entire problem to a system that is expertly designed to solve it. PostgreSQL ensures that once our COMMIT statement succeeds, the data is durably stored and will survive a crash. How does it do this? With its own, highly optimized, internal commit log, called the Write-Ahead Log (WAL).

This is an interesting parallel: we are using a database (which itself is built on a commit log) to implement our application-level commit log.

The Trade-offs:

• Pros:

  • Durability & ACID: We get full ACID guarantees out of the box. No need to implement fsync, CRCs, or recovery logic.
  • Replication & Backups: We can leverage standard PostgreSQL tools for high availability and disaster recovery.
  • Operational Simplicity: If you already run a PostgreSQL cluster, adding another table is operationally much simpler than managing a new distributed file storage system.

• Cons:

  • Performance: A general-purpose database will likely have lower raw throughput for append-only writes compared to a specialized file-based system like the one we built in the last lesson. There is overhead from SQL parsing, transaction management, and indexing.
  • Coupling: It couples our broker's performance and availability to the database cluster.

For many use cases, the reliability and operational benefits of using a database like PostgreSQL far outweigh the potential performance cost.

Conclusion

Excellent work. You have now created a second, fully functional storage backend for our message broker, this time using PostgreSQL. Because both Log and PostgresLog implement the CommitLog interface, our broker's core logic can use either one without any code changes.

Key Takeaways:

• The Repository Pattern is a powerful tool for decoupling application logic from data storage details.
• A relational database like PostgreSQL can serve as a robust backend for a commit log, modeling partitions and offsets with a simple table schema.
• Using SELECT ... FOR UPDATE within a transaction is a key technique for safely managing sequential writes in a concurrent environment.
• By building on PostgreSQL, we inherit its sophisticated Write-Ahead Log (WAL), giving us strong durability and consistency guarantees for free.

Next Up

In the next lesson, we will implement the logic for creating and deleting topics. This "topic manager" will be responsible for instantiating the appropriate CommitLog implementation (file-based or PostgreSQL) for each new partition, truly bringing our abstract design to life.