Track WAL segments and compact acknowledged files

Make WAL writes and replays segment-aware and compact acknowledged segments. Introduce bufferedSample to RingBuffer to carry walSegment IDs, add PushWithWALSegment/DrainAllWithWALSegments, and update buffer/peek semantics. Extend WAL with WriteSampleWithSegment/WriteRecordWithSegment and ReplaySamplesWithSegments (returns segment indices) and perform compaction in AcknowledgeThrough by removing fully acknowledged segment files. Update Agent to write samples with segment IDs, preserve segment metadata through the buffer, and acknowledge/compact WAL on successful exports. Update README/docs and add tests for replay-with-segments, acknowledgement compaction, and agent WAL acknowledgement behavior.

Author: vagnercazarotto

README.md

@@ -232,7 +232,7 @@ When zero values are provided in `pulse.Config`, defaults are applied:
 Optional WAL:
 
 - Set `Config.WAL` with a valid directory to enable disk persistence and replay.
-- Current implementation focuses on durable segmented writes and startup replay; segment cleanup remains conservative.
+- Successful exports acknowledge WAL segments and trigger compaction of acknowledged files.
 
 ## Runtime metrics currently collected
 

agent.go

@@ -64,14 +64,14 @@ func (a *Agent) Start() error {
 		}
 		a.wal = w
 
-		replayed, _, err := a.wal.ReplaySamples()
+		replayed, _, err := a.wal.ReplaySamplesWithSegments()
 		if err != nil {
 			_ = a.wal.Close()
 			a.wal = nil
 			return err
 		}
-		for _, s := range replayed {
-			a.buffer.Push(s)
+		for _, item := range replayed {
+			a.buffer.PushWithWALSegment(item.Sample, item.Segment)
 		}
 	}
 
@@ -144,10 +144,14 @@ func (a *Agent) collectLoop() {
 			return
 		case <-ticker.C:
 			s := a.collectSample()
-			a.buffer.Push(s)
+			walSegment := -1
 			if a.wal != nil {
-				_ = a.wal.WriteSample(s)
+				seg, err := a.wal.WriteSampleWithSegment(s)
+				if err == nil {
+					walSegment = seg
+				}
 			}
+			a.buffer.PushWithWALSegment(s, walSegment)
 		}
 	}
 }
@@ -231,19 +235,32 @@ func collectRuntimeValues() map[string]float64 {
 }
 
 func (a *Agent) flushExports() {
-	samples := a.buffer.DrainAll()
-	if len(samples) == 0 || len(a.cfg.Exporters) == 0 {
+	entries := a.buffer.DrainAllWithWALSegments()
+	if len(entries) == 0 || len(a.cfg.Exporters) == 0 {
 		return
 	}
 
+	samples := make([]Sample, 0, len(entries))
+	maxWALSegment := -1
+	for _, entry := range entries {
+		samples = append(samples, entry.sample)
+		if entry.walSegment > maxWALSegment {
+			maxWALSegment = entry.walSegment
+		}
+	}
+
 	for _, exp := range a.cfg.Exporters {
 		if err := a.exportWithRetry(exp, samples); err != nil {
-			for _, s := range samples {
-				a.buffer.Push(s)
+			for _, entry := range entries {
+				a.buffer.PushWithWALSegment(entry.sample, entry.walSegment)
 			}
 			return
 		}
 	}
+
+	if a.wal != nil && maxWALSegment >= 0 {
+		a.wal.AcknowledgeThrough(maxWALSegment)
+	}
 }
 
 func (a *Agent) exportWithRetry(exp Exporter, samples []Sample) error {

agent_test.go

@@ -234,3 +234,62 @@ func TestFlushExportsNonRetryableFailsFast(t *testing.T) {
 		t.Fatalf("expected sample to be requeued")
 	}
 }
+
+func TestFlushExportsAcknowledgesWALOnSuccess(t *testing.T) {
+	dir := t.TempDir()
+	w, err := NewWAL(WALConfig{Dir: dir, SegmentSize: 128, SyncEvery: 1})
+	if err != nil {
+		t.Fatalf("new wal: %v", err)
+	}
+	defer w.Close()
+
+	rec := &recordingExporter{}
+	a := New(Config{
+		Exporters:            []Exporter{rec},
+		ExportMaxRetries:     1,
+		ExportBackoffInitial: 1 * time.Millisecond,
+		ExportBackoffMax:     2 * time.Millisecond,
+		ExportBackoffJitter:  0,
+	})
+	a.ctx = context.Background()
+	a.wal = w
+
+	maxSegment := -1
+	for i := 0; i < 20; i++ {
+		s := Sample{
+			Timestamp: time.Now().UTC(),
+			Values: map[string]float64{
+				"v":   float64(i),
+				"pad": float64(i * 1000),
+			},
+		}
+		seg, err := w.WriteSampleWithSegment(s)
+		if err != nil {
+			t.Fatalf("write sample %d: %v", i, err)
+		}
+		if seg > maxSegment {
+			maxSegment = seg
+		}
+		a.buffer.PushWithWALSegment(s, seg)
+	}
+
+	if maxSegment < 1 {
+		t.Fatalf("expected multiple WAL segments, got max %d", maxSegment)
+	}
+
+	a.flushExports()
+
+	if w.acknowledgedThrough != maxSegment {
+		t.Fatalf("expected acknowledgedThrough=%d, got %d", maxSegment, w.acknowledgedThrough)
+	}
+
+	segments, err := listSegments(dir)
+	if err != nil {
+		t.Fatalf("list segments: %v", err)
+	}
+	for _, seg := range segments {
+		if idx := segmentIndex(seg); idx < maxSegment {
+			t.Fatalf("expected old segment compacted, found %s", seg)
+		}
+	}
+}

buffer.go

@@ -11,61 +11,87 @@ var ErrInvalidBufferCapacity = errors.New("pulse: buffer capacity must be greate
 // RingBuffer is a fixed-size, thread-safe circular buffer for samples.
 type RingBuffer struct {
 	mu   sync.Mutex
-	data []Sample
+	data []bufferedSample
 	head int
 	size int
 
 	overflow atomic.Uint64
 }
 
+type bufferedSample struct {
+	sample     Sample
+	walSegment int
+}
+
 // NewRingBuffer creates a new ring buffer with fixed capacity.
 func NewRingBuffer(capacity int) (*RingBuffer, error) {
 	if capacity <= 0 {
 		return nil, ErrInvalidBufferCapacity
 	}
-	return &RingBuffer{data: make([]Sample, capacity)}, nil
+	return &RingBuffer{data: make([]bufferedSample, capacity)}, nil
 }
 
 // Push inserts a sample. If full, the oldest sample is evicted.
 func (b *RingBuffer) Push(s Sample) {
+	b.PushWithWALSegment(s, -1)
+}
+
+// PushWithWALSegment inserts a sample and tracks the WAL segment id when known.
+func (b *RingBuffer) PushWithWALSegment(s Sample, walSegment int) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
 	if len(b.data) == 0 {
 		return
 	}
+	entry := bufferedSample{sample: cloneSample(s), walSegment: walSegment}
 
 	if b.size == len(b.data) {
-		b.data[b.head] = cloneSample(s)
+		b.data[b.head] = entry
 		b.head = (b.head + 1) % len(b.data)
 		b.overflow.Add(1)
 		return
 	}
 
 	tail := (b.head + b.size) % len(b.data)
-	b.data[tail] = cloneSample(s)
+	b.data[tail] = entry
 	b.size++
 }
 
-// DrainAll returns all samples in insertion order and clears the buffer.
-func (b *RingBuffer) DrainAll() []Sample {
+// DrainAllWithWALSegments returns all buffered entries and clears the buffer.
+func (b *RingBuffer) DrainAllWithWALSegments() []bufferedSample {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
 	if b.size == 0 {
 		return nil
 	}
 
-	out := make([]Sample, 0, b.size)
+	out := make([]bufferedSample, 0, b.size)
 	for i := 0; i < b.size; i++ {
 		idx := (b.head + i) % len(b.data)
-		out = append(out, cloneSample(b.data[idx]))
+		entry := b.data[idx]
+		entry.sample = cloneSample(entry.sample)
+		out = append(out, entry)
 	}
 	b.head = 0
 	b.size = 0
 	return out
 }
 
+// DrainAll returns all samples in insertion order and clears the buffer.
+func (b *RingBuffer) DrainAll() []Sample {
+	entries := b.DrainAllWithWALSegments()
+	if len(entries) == 0 {
+		return nil
+	}
+	out := make([]Sample, 0, len(entries))
+	for _, entry := range entries {
+		out = append(out, entry.sample)
+	}
+	return out
+}
+
 // Peek returns up to n samples in insertion order without removing them.
 func (b *RingBuffer) Peek(n int) []Sample {
 	b.mu.Lock()
@@ -81,7 +107,7 @@ func (b *RingBuffer) Peek(n int) []Sample {
 	out := make([]Sample, 0, n)
 	for i := 0; i < n; i++ {
 		idx := (b.head + i) % len(b.data)
-		out = append(out, cloneSample(b.data[idx]))
+		out = append(out, cloneSample(b.data[idx].sample))
 	}
 	return out
 }

docs/current-reference.html

@@ -69,7 +69,7 @@ <h2>WAL (persistence)</h2>
         <li>Format: length prefix + payload + crc32.</li>
         <li>Replay in order; partial tail is truncated to the last valid record.</li>
         <li>Rotation: 64 MiB segments, 512 MiB total cap.</li>
-        <li>The current build prioritizes durable append and startup replay; segment cleanup remains conservative.</li>
+        <li>Successful exports acknowledge WAL segments and compact acknowledged files.</li>
       </ul>
     </section>