dailz
fe2d4fc5f0
- wal/commit_queue.go: bounded buffered channel for write requests - wal/writer.go: single-goroutine main loop implementing 11-step write flow with group commit, sequence allocation, MemTable publish/abort, write-stopped - wal/recovery.go: BatchReplayer interface, ReplayBatch, ReplaySegmentFile, RecoverFromSegments with fragment reassembly and tail corruption handling - Comprehensive tests for all modules, all pass with -race
59 lines
1.4 KiB
Go
59 lines
1.4 KiB
Go
package wal
|
|
|
|
// WriteResult is sent to the caller when the write completes or fails.
|
|
type WriteResult struct {
|
|
Sequence uint64
|
|
Err error
|
|
}
|
|
|
|
// CommitRequest represents a single write request submitted by a caller.
|
|
type CommitRequest struct {
|
|
Entries []*WalEntry
|
|
Result chan WriteResult
|
|
}
|
|
|
|
// CommitQueue is a bounded buffered channel for write requests.
|
|
type CommitQueue struct {
|
|
ch chan *CommitRequest
|
|
}
|
|
|
|
// NewCommitQueue creates a bounded queue for commit requests.
|
|
func NewCommitQueue(capacity int) *CommitQueue {
|
|
if capacity < 1 {
|
|
capacity = 1
|
|
}
|
|
return &CommitQueue{ch: make(chan *CommitRequest, capacity)}
|
|
}
|
|
|
|
// Submit creates a commit request, submits it to the queue, and returns it.
|
|
// The send blocks when the queue is full, preserving backpressure.
|
|
func (cq *CommitQueue) Submit(entries []*WalEntry) *CommitRequest {
|
|
req := &CommitRequest{
|
|
Entries: entries,
|
|
Result: make(chan WriteResult, 1),
|
|
}
|
|
cq.ch <- req
|
|
return req
|
|
}
|
|
|
|
// Collect drains all currently pending requests from the queue.
|
|
func (cq *CommitQueue) Collect() []*CommitRequest {
|
|
var requests []*CommitRequest
|
|
for {
|
|
select {
|
|
case req, ok := <-cq.ch:
|
|
if !ok {
|
|
return requests
|
|
}
|
|
requests = append(requests, req)
|
|
default:
|
|
return requests
|
|
}
|
|
}
|
|
}
|
|
|
|
// Close closes the queue. Callers must stop submitting before Close.
|
|
func (cq *CommitQueue) Close() {
|
|
close(cq.ch)
|
|
}
|