Delay oplog deletion depending on active readers

[lalinsky]

I'd like to have a mechanism to ensuring that when a new IndexReader is acquired, we will always have oplog files needed to catch up from the last commit in the reader, if we need to make a snapshot of it.

Currently, we truncate oplog when we commit a new file segment. Instead of that, we should have some kind of tracker for the smallest commit ID we need to have oplog entries for.

@claude think about this and propose a solution

[claude]

Claude finished @lalinsky's task —— View job

---

Oplog deletion and reader tracking solution

I've analyzed the current implementation and propose the following solution for delaying oplog deletion based on active readers.

Todo List

• Read CLAUDE.md for context
• Examine oplog implementation (src/Oplog.zig)
• Examine IndexReader and snapshot functionality
• Understand current truncation logic
• Design solution for tracking active readers
• Propose implementation approach

Problem Analysis

Currently in src/Index.zig:180, when a checkpoint creates a new file segment, the oplog is immediately truncated up to that segment's last commit ID. This can delete oplog entries that active IndexReaders might still need to catch up from their snapshot point.

Proposed Solution

1. Reader Tracking Structure
Add a ReaderTracker component to track active readers and their minimum required commit IDs:

const ReaderTracker = struct {
    mutex: std.Thread.Mutex = .{},
    active_readers: std.HashMap(u64, ReaderInfo, std.hash_map.DefaultContext(u64), std.hash_map.default_max_load_percentage),
    next_reader_id: u64 = 1,
    
    const ReaderInfo = struct {
        min_commit_id: u64,
        acquired_at: i64, // timestamp for debugging/cleanup
    };
    
    pub fn registerReader(self: *ReaderTracker, min_commit_id: u64) u64 {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        const reader_id = self.next_reader_id;
        self.next_reader_id += 1;
        
        self.active_readers.put(reader_id, .{
            .min_commit_id = min_commit_id,
            .acquired_at = std.time.timestamp(),
        }) catch unreachable; // handle appropriately
        
        return reader_id;
    }
    
    pub fn unregisterReader(self: *ReaderTracker, reader_id: u64) void {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        _ = self.active_readers.remove(reader_id);
    }
    
    pub fn getMinRequiredCommitId(self: *ReaderTracker) ?u64 {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        var min_commit_id: ?u64 = null;
        var it = self.active_readers.valueIterator();
        while (it.next()) |reader_info| {
            if (min_commit_id == null or reader_info.min_commit_id < min_commit_id.?) {
                min_commit_id = reader_info.min_commit_id;
            }
        }
        return min_commit_id;
    }
};

2. Enhanced IndexReader
Modify IndexReader to track its ID and minimum commit requirement:

// In IndexReader.zig
reader_id: u64,
min_commit_id: u64,
index: *Index, // reference back to parent for cleanup

3. Modified Index Operations
Update Index.acquireReader() and Index.releaseReader():

// In Index.zig
reader_tracker: ReaderTracker,

pub fn acquireReader(self: *Self) !IndexReader {
    self.segments_lock.lockShared();
    defer self.segments_lock.unlockShared();
    
    const reader_snapshot = IndexReader{
        .file_segments = self.file_segments.segments.acquire(),
        .memory_segments = self.memory_segments.segments.acquire(),
    };
    
    // Determine minimum commit ID this reader might need
    const min_commit_id = reader_snapshot.getVersion();
    const reader_id = self.reader_tracker.registerReader(min_commit_id);
    
    return IndexReader{
        .file_segments = reader_snapshot.file_segments,
        .memory_segments = reader_snapshot.memory_segments,
        .reader_id = reader_id,
        .min_commit_id = min_commit_id,
        .index = self,
    };
}

pub fn releaseReader(self: *Self, reader: *IndexReader) void {
    self.reader_tracker.unregisterReader(reader.reader_id);
    MemorySegmentList.destroySegments(self.allocator, &reader.memory_segments);
    FileSegmentList.destroySegments(self.allocator, &reader.file_segments);
}

4. Safe Oplog Truncation
Modify the truncation logic in checkpoint():

// In Index.zig checkpoint function, replace line 180:
const safe_truncate_point = if (self.reader_tracker.getMinRequiredCommitId()) |min_required|
    @min(target.value.info.getLastCommitId(), min_required - 1)
else
    target.value.info.getLastCommitId();

defer self.oplog.truncate(safe_truncate_point) catch |err| {
    log.warn("failed to truncate oplog: {}", .{err});
};

Benefits

• Safety: Guarantees oplog availability for snapshot restoration
• Automatic cleanup: Readers are tracked and cleaned up automatically
• Performance: Minimal overhead with HashMap lookup
• Backward compatibility: No changes to existing API surface

Implementation Notes

• Add ReaderTracker field to Index struct
• Initialize tracker in Index.init()
• Clean up tracker in Index.deinit()
• Consider adding metrics for tracking reader count and oplog retention

This solution ensures that oplog files remain available as long as any IndexReader that might need them for snapshot restoration is still active.

[lalinsky]

@claude what about adding some kind of "replication slot" concept to oplog, so the main index would keep advancing one slot, and we can have a new parameter in acquireReader to optionally get a slot as well.

[claude]

Claude finished @lalinsky's task —— View job

---

Replication slot concept analysis

Todo List

• Examine current oplog implementation for slot mechanics
• Analyze IndexReader acquisition flow
• Design replication slot system
• Compare with previous HashMap approach
• Provide design recommendation

Current Implementation Analysis

The oplog currently works with a simple linear commit ID system:

• last_commit_id tracks the latest commit (src/Oplog.zig:36)
• New commits increment sequentially (src/Index.zig:476)
• Truncation happens at checkpoint using target.value.info.getLastCommitId() (src/Index.zig:180)
• IndexReader snapshots capture current version via getVersion() (src/IndexReader.zig:104-112)

Replication Slot Design

The replication slot concept is excellent! It's simpler and more PostgreSQL-like than the HashMap approach. Here's the refined design:

Core Concept:

• Main index advances the "main slot" with each commit
• Readers can optionally acquire a "replication slot" that tracks their minimum required commit ID
• Oplog truncation respects the oldest active replication slot

Implementation:

// In Oplog.zig - add replication slot tracking
pub const ReplicationSlot = struct {
    commit_id: std.atomic.Value(u64),
    in_use: std.atomic.Value(bool),
    
    pub fn init(commit_id: u64) ReplicationSlot {
        return .{
            .commit_id = std.atomic.Value(u64).init(commit_id),
            .in_use = std.atomic.Value(bool).init(true),
        };
    }
    
    pub fn release(self: *ReplicationSlot) void {
        self.in_use.store(false, .release);
    }
    
    pub fn advance(self: *ReplicationSlot, new_commit_id: u64) void {
        self.commit_id.store(new_commit_id, .release);
    }
};

// In Oplog.zig - add slot management
replication_slots: std.ArrayList(*ReplicationSlot),
slots_lock: std.Thread.Mutex = .{},

pub fn acquireReplicationSlot(self: *Self, commit_id: u64) !*ReplicationSlot {
    const slot = try self.allocator.create(ReplicationSlot);
    slot.* = ReplicationSlot.init(commit_id);
    
    self.slots_lock.lock();
    defer self.slots_lock.unlock();
    
    try self.replication_slots.append(slot);
    return slot;
}

pub fn releaseReplicationSlot(self: *Self, slot: *ReplicationSlot) void {
    slot.release();
    // Cleanup will happen during next truncation
}

fn getMinActiveSlotCommitId(self: *Self) ?u64 {
    self.slots_lock.lock();
    defer self.slots_lock.unlock();
    
    var min_commit_id: ?u64 = null;
    var i: usize = 0;
    
    // Clean up released slots while finding minimum
    while (i < self.replication_slots.items.len) {
        const slot = self.replication_slots.items[i];
        if (!slot.in_use.load(.acquire)) {
            self.allocator.destroy(slot);
            _ = self.replication_slots.swapRemove(i);
        } else {
            const commit_id = slot.commit_id.load(.acquire);
            if (min_commit_id == null or commit_id < min_commit_id.?) {
                min_commit_id = commit_id;
            }
            i += 1;
        }
    }
    
    return min_commit_id;
}

Modified Index API:

// In Index.zig - enhanced acquireReader with optional slot parameter
pub fn acquireReader(self: *Self) !IndexReader { 
    return self.acquireReaderWithSlot(false);
}

pub fn acquireReaderWithSlot(self: *Self, use_slot: bool) !IndexReader {
    self.segments_lock.lockShared();
    defer self.segments_lock.unlockShared();
    
    const current_version = self.getCurrentVersion();
    const slot = if (use_slot) try self.oplog.acquireReplicationSlot(current_version) else null;
    
    return IndexReader{
        .file_segments = self.file_segments.segments.acquire(),
        .memory_segments = self.memory_segments.segments.acquire(),
        .replication_slot = slot,
    };
}

pub fn releaseReader(self: *Self, reader: *IndexReader) void {
    if (reader.replication_slot) |slot| {
        self.oplog.releaseReplicationSlot(slot);
    }
    MemorySegmentList.destroySegments(self.allocator, &reader.memory_segments);
    FileSegmentList.destroySegments(self.allocator, &reader.file_segments);
}

Safe Truncation:

// In Index.zig checkpoint() - replace line 180:
const safe_truncate_point = if (self.oplog.getMinActiveSlotCommitId()) |min_slot_commit|
    @min(target.value.info.getLastCommitId(), min_slot_commit - 1) 
else 
    target.value.info.getLastCommitId();

defer self.oplog.truncate(safe_truncate_point) catch |err| {
    log.warn("failed to truncate oplog: {}", .{err});
};

Comparison: Replication Slots vs HashMap

Replication Slots (Recommended):

• ✅ PostgreSQL-familiar: Well-understood concept
• ✅ Optional: acquireReader() unchanged, acquireReaderWithSlot(true) for slot behavior
• ✅ Lock-free reads: Atomic operations for slot access
• ✅ Auto-cleanup: Released slots cleaned up during truncation
• ✅ Simpler: No reader ID management needed

HashMap Approach:

• ❌ Always tracking: All readers tracked, even if not needed
• ❌ More complex: Reader ID generation and cleanup required
• ❌ Lock contention: Mutex for every reader operation

Recommendation

Use the replication slot approach - it's cleaner, optional, and follows established database patterns. The key insight is making slot acquisition optional via acquireReaderWithSlot(use_slot: bool), keeping the existing API unchanged while enabling fine-grained oplog retention control when needed.

---