Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d8fe3f5a3c |
@@ -0,0 +1,331 @@
|
|||||||
|
package batching
|
||||||
|
|
||||||
|
import (
|
||||||
|
"fmt"
|
||||||
|
"sync"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
// GitOp represents a git operation to be batched
|
||||||
|
type GitOp struct {
|
||||||
|
OpType string // "commit", "push", "merge"
|
||||||
|
Branch string
|
||||||
|
Message string
|
||||||
|
Files []string
|
||||||
|
Timestamp time.Time
|
||||||
|
ID string
|
||||||
|
}
|
||||||
|
|
||||||
|
// GitBatch represents a batch of git operations
|
||||||
|
type GitBatch struct {
|
||||||
|
ID string
|
||||||
|
Operations []*GitOp
|
||||||
|
CreatedAt time.Time
|
||||||
|
ExecutedAt time.Time
|
||||||
|
Status string // "pending", "executing", "completed", "failed"
|
||||||
|
Error error
|
||||||
|
}
|
||||||
|
|
||||||
|
// GitBatcher batches git operations for efficient execution
|
||||||
|
type GitBatcher struct {
|
||||||
|
mu sync.RWMutex
|
||||||
|
queue []*GitOp
|
||||||
|
maxBatchSize int
|
||||||
|
maxBatchAge time.Duration
|
||||||
|
lastFlushTime time.Time
|
||||||
|
executedBatches []*GitBatch
|
||||||
|
pendingBatches []*GitBatch
|
||||||
|
stats *BatchStats
|
||||||
|
flushChan chan struct{}
|
||||||
|
stopChan chan struct{}
|
||||||
|
}
|
||||||
|
|
||||||
|
// BatchStats tracks batching statistics
|
||||||
|
type BatchStats struct {
|
||||||
|
TotalOps int
|
||||||
|
TotalBatches int
|
||||||
|
AvgOpsPerBatch float64
|
||||||
|
NetworkSavings int // Estimated network round trips saved
|
||||||
|
TotalExecuteTime time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewGitBatcher creates a new git batcher
|
||||||
|
func NewGitBatcher(maxBatchSize int, maxBatchAge time.Duration) *GitBatcher {
|
||||||
|
if maxBatchSize <= 0 {
|
||||||
|
maxBatchSize = 10
|
||||||
|
}
|
||||||
|
if maxBatchAge <= 0 {
|
||||||
|
maxBatchAge = 5 * time.Second
|
||||||
|
}
|
||||||
|
|
||||||
|
return &GitBatcher{
|
||||||
|
queue: make([]*GitOp, 0),
|
||||||
|
maxBatchSize: maxBatchSize,
|
||||||
|
maxBatchAge: maxBatchAge,
|
||||||
|
lastFlushTime: time.Now(),
|
||||||
|
executedBatches: make([]*GitBatch, 0),
|
||||||
|
pendingBatches: make([]*GitBatch, 0),
|
||||||
|
stats: &BatchStats{
|
||||||
|
TotalOps: 0,
|
||||||
|
TotalBatches: 0,
|
||||||
|
},
|
||||||
|
flushChan: make(chan struct{}, 1),
|
||||||
|
stopChan: make(chan struct{}),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Enqueue adds a git operation to the queue
|
||||||
|
func (gb *GitBatcher) Enqueue(op *GitOp) {
|
||||||
|
if op == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
op.Timestamp = time.Now()
|
||||||
|
|
||||||
|
gb.mu.Lock()
|
||||||
|
defer gb.mu.Unlock()
|
||||||
|
|
||||||
|
gb.queue = append(gb.queue, op)
|
||||||
|
gb.stats.TotalOps++
|
||||||
|
|
||||||
|
// Auto-flush if batch is full
|
||||||
|
if len(gb.queue) >= gb.maxBatchSize {
|
||||||
|
gb.flushLocked()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// flushLocked creates a batch from queued operations (must be called with lock held)
|
||||||
|
func (gb *GitBatcher) flushLocked() {
|
||||||
|
if len(gb.queue) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
batch := &GitBatch{
|
||||||
|
ID: fmt.Sprintf("batch-%d", gb.stats.TotalBatches),
|
||||||
|
Operations: make([]*GitOp, len(gb.queue)),
|
||||||
|
CreatedAt: time.Now(),
|
||||||
|
Status: "pending",
|
||||||
|
}
|
||||||
|
|
||||||
|
copy(batch.Operations, gb.queue)
|
||||||
|
|
||||||
|
gb.pendingBatches = append(gb.pendingBatches, batch)
|
||||||
|
gb.queue = make([]*GitOp, 0)
|
||||||
|
gb.lastFlushTime = time.Now()
|
||||||
|
gb.stats.TotalBatches++
|
||||||
|
}
|
||||||
|
|
||||||
|
// Flush manually flushes the current batch
|
||||||
|
func (gb *GitBatcher) Flush() {
|
||||||
|
gb.mu.Lock()
|
||||||
|
defer gb.mu.Unlock()
|
||||||
|
|
||||||
|
gb.flushLocked()
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetPendingBatch returns the next pending batch without removing it
|
||||||
|
func (gb *GitBatcher) GetPendingBatch() *GitBatch {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
if len(gb.pendingBatches) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return gb.pendingBatches[0]
|
||||||
|
}
|
||||||
|
|
||||||
|
// MarkBatchExecuting marks a batch as executing
|
||||||
|
func (gb *GitBatcher) MarkBatchExecuting(batchID string) {
|
||||||
|
gb.mu.Lock()
|
||||||
|
defer gb.mu.Unlock()
|
||||||
|
|
||||||
|
for _, batch := range gb.pendingBatches {
|
||||||
|
if batch.ID == batchID {
|
||||||
|
batch.Status = "executing"
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// MarkBatchCompleted marks a batch as completed and removes from pending
|
||||||
|
func (gb *GitBatcher) MarkBatchCompleted(batchID string) {
|
||||||
|
gb.mu.Lock()
|
||||||
|
defer gb.mu.Unlock()
|
||||||
|
|
||||||
|
var idx int
|
||||||
|
var found *GitBatch
|
||||||
|
for i, batch := range gb.pendingBatches {
|
||||||
|
if batch.ID == batchID {
|
||||||
|
idx = i
|
||||||
|
found = batch
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if found != nil {
|
||||||
|
found.Status = "completed"
|
||||||
|
found.ExecutedAt = time.Now()
|
||||||
|
|
||||||
|
// Move to executed batches
|
||||||
|
gb.executedBatches = append(gb.executedBatches, found)
|
||||||
|
|
||||||
|
// Remove from pending
|
||||||
|
gb.pendingBatches = append(gb.pendingBatches[:idx], gb.pendingBatches[idx+1:]...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// MarkBatchFailed marks a batch as failed with an error
|
||||||
|
func (gb *GitBatcher) MarkBatchFailed(batchID string, err error) {
|
||||||
|
gb.mu.Lock()
|
||||||
|
defer gb.mu.Unlock()
|
||||||
|
|
||||||
|
var found *GitBatch
|
||||||
|
for _, batch := range gb.pendingBatches {
|
||||||
|
if batch.ID == batchID {
|
||||||
|
found = batch
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if found != nil {
|
||||||
|
found.Status = "failed"
|
||||||
|
found.Error = err
|
||||||
|
found.ExecutedAt = time.Now()
|
||||||
|
|
||||||
|
// Keep in pending (for retry logic)
|
||||||
|
// Could also move to failed queue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// QueueSize returns the current queue size
|
||||||
|
func (gb *GitBatcher) QueueSize() int {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
return len(gb.queue)
|
||||||
|
}
|
||||||
|
|
||||||
|
// PendingBatchCount returns the number of pending batches
|
||||||
|
func (gb *GitBatcher) PendingBatchCount() int {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
return len(gb.pendingBatches)
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetStats returns batching statistics
|
||||||
|
func (gb *GitBatcher) GetStats() *BatchStats {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
stats := *gb.stats
|
||||||
|
if stats.TotalBatches > 0 {
|
||||||
|
stats.AvgOpsPerBatch = float64(stats.TotalOps) / float64(stats.TotalBatches)
|
||||||
|
// Estimated savings: each batch saves (ops-1) round trips
|
||||||
|
stats.NetworkSavings = stats.TotalOps - stats.TotalBatches
|
||||||
|
}
|
||||||
|
|
||||||
|
return &stats
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetExecutedBatches returns all executed batches
|
||||||
|
func (gb *GitBatcher) GetExecutedBatches() []*GitBatch {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
result := make([]*GitBatch, len(gb.executedBatches))
|
||||||
|
copy(result, gb.executedBatches)
|
||||||
|
|
||||||
|
return result
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetBatchByID returns a specific batch by ID
|
||||||
|
func (gb *GitBatcher) GetBatchByID(batchID string) *GitBatch {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
for _, batch := range gb.pendingBatches {
|
||||||
|
if batch.ID == batchID {
|
||||||
|
return batch
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, batch := range gb.executedBatches {
|
||||||
|
if batch.ID == batchID {
|
||||||
|
return batch
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// TimeSinceLastFlush returns time since last flush
|
||||||
|
func (gb *GitBatcher) TimeSinceLastFlush() time.Duration {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
return time.Since(gb.lastFlushTime)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ShouldFlush checks if batch should be flushed based on age
|
||||||
|
func (gb *GitBatcher) ShouldFlush() bool {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
if len(gb.queue) == 0 {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
return time.Since(gb.lastFlushTime) >= gb.maxBatchAge
|
||||||
|
}
|
||||||
|
|
||||||
|
// Clear clears all pending operations and batches
|
||||||
|
func (gb *GitBatcher) Clear() {
|
||||||
|
gb.mu.Lock()
|
||||||
|
defer gb.mu.Unlock()
|
||||||
|
|
||||||
|
gb.queue = make([]*GitOp, 0)
|
||||||
|
gb.pendingBatches = make([]*GitBatch, 0)
|
||||||
|
gb.executedBatches = make([]*GitBatch, 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetQueuedOps returns a copy of queued operations
|
||||||
|
func (gb *GitBatcher) GetQueuedOps() []*GitOp {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
ops := make([]*GitOp, len(gb.queue))
|
||||||
|
copy(ops, gb.queue)
|
||||||
|
|
||||||
|
return ops
|
||||||
|
}
|
||||||
|
|
||||||
|
// CalculateNetworkSavings calculates estimated network round trips saved
|
||||||
|
func (gb *GitBatcher) CalculateNetworkSavings() int {
|
||||||
|
gb.mu.RLock()
|
||||||
|
defer gb.mu.RUnlock()
|
||||||
|
|
||||||
|
totalSavings := 0
|
||||||
|
// Each batch of N operations saves N-1 round trips
|
||||||
|
for _, batch := range gb.executedBatches {
|
||||||
|
if len(batch.Operations) > 1 {
|
||||||
|
totalSavings += len(batch.Operations) - 1
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return totalSavings
|
||||||
|
}
|
||||||
|
|
||||||
|
// GetBatchInfo returns human-readable batch information
|
||||||
|
func (batch *GitBatch) GetInfo() map[string]interface{} {
|
||||||
|
return map[string]interface{}{
|
||||||
|
"id": batch.ID,
|
||||||
|
"status": batch.Status,
|
||||||
|
"op_count": len(batch.Operations),
|
||||||
|
"created_at": batch.CreatedAt,
|
||||||
|
"executed_at": batch.ExecutedAt,
|
||||||
|
"duration": batch.ExecutedAt.Sub(batch.CreatedAt),
|
||||||
|
"error": batch.Error,
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,355 @@
|
|||||||
|
package batching
|
||||||
|
|
||||||
|
import (
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestNewGitBatcher(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
assert.NotNil(t, batcher)
|
||||||
|
assert.Equal(t, 0, batcher.QueueSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnqueueOperation(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
Message: "Add feature",
|
||||||
|
Files: []string{"file1.go"},
|
||||||
|
}
|
||||||
|
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
assert.Equal(t, 1, batcher.QueueSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnqueueMultipleOps(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
Message: "Commit",
|
||||||
|
}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
|
||||||
|
assert.Equal(t, 5, batcher.QueueSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAutoFlushOnMaxBatchSize(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(5, 10*time.Second)
|
||||||
|
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
Message: "Commit",
|
||||||
|
}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
|
||||||
|
// After 5 ops, should auto-flush
|
||||||
|
assert.Equal(t, 0, batcher.QueueSize())
|
||||||
|
assert.Equal(t, 1, batcher.PendingBatchCount())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestManualFlush(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
Message: "Commit",
|
||||||
|
}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
assert.Equal(t, 1, batcher.QueueSize())
|
||||||
|
|
||||||
|
batcher.Flush()
|
||||||
|
assert.Equal(t, 0, batcher.QueueSize())
|
||||||
|
assert.Equal(t, 1, batcher.PendingBatchCount())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetPendingBatch(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
Message: "Commit",
|
||||||
|
}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
assert.NotNil(t, batch)
|
||||||
|
assert.Equal(t, 1, len(batch.Operations))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMarkBatchExecuting(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
batcher.MarkBatchExecuting(batch.ID)
|
||||||
|
|
||||||
|
updated := batcher.GetBatchByID(batch.ID)
|
||||||
|
assert.Equal(t, "executing", updated.Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMarkBatchCompleted(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
batcher.MarkBatchCompleted(batch.ID)
|
||||||
|
|
||||||
|
executed := batcher.GetExecutedBatches()
|
||||||
|
assert.Equal(t, 1, len(executed))
|
||||||
|
assert.Equal(t, "completed", executed[0].Status)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMarkBatchFailed(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
testErr := assert.AnError
|
||||||
|
batcher.MarkBatchFailed(batch.ID, testErr)
|
||||||
|
|
||||||
|
failed := batcher.GetBatchByID(batch.ID)
|
||||||
|
assert.Equal(t, "failed", failed.Status)
|
||||||
|
assert.Error(t, failed.Error)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetStats(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(5, 5*time.Second)
|
||||||
|
|
||||||
|
// Add 10 ops (will create 2 batches of 5 each)
|
||||||
|
for i := 0; i < 10; i++ {
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
|
||||||
|
stats := batcher.GetStats()
|
||||||
|
assert.Equal(t, 10, stats.TotalOps)
|
||||||
|
assert.Equal(t, 2, stats.TotalBatches)
|
||||||
|
assert.Equal(t, 5.0, stats.AvgOpsPerBatch)
|
||||||
|
// 10 ops in 2 batches saves 8 round trips (5-1 + 5-1)
|
||||||
|
assert.Equal(t, 8, stats.NetworkSavings)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestQueueSize(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
|
||||||
|
assert.Equal(t, 1, batcher.QueueSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestPendingBatchCount(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
assert.Equal(t, 1, batcher.PendingBatchCount())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetExecutedBatches(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
// Create and execute batches
|
||||||
|
for i := 0; i < 2; i++ {
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
batcher.MarkBatchCompleted(batch.ID)
|
||||||
|
}
|
||||||
|
|
||||||
|
executed := batcher.GetExecutedBatches()
|
||||||
|
assert.Equal(t, 2, len(executed))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTimeSinceLastFlush(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
time.Sleep(100 * time.Millisecond)
|
||||||
|
elapsed := batcher.TimeSinceLastFlush()
|
||||||
|
|
||||||
|
assert.Greater(t, elapsed, 50*time.Millisecond)
|
||||||
|
assert.Less(t, elapsed, 200*time.Millisecond)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestShouldFlush(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(100, 100*time.Millisecond)
|
||||||
|
|
||||||
|
// Empty queue should not flush
|
||||||
|
assert.False(t, batcher.ShouldFlush())
|
||||||
|
|
||||||
|
// Enqueue but not old enough
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
assert.False(t, batcher.ShouldFlush())
|
||||||
|
|
||||||
|
// Wait for age to exceed max age
|
||||||
|
time.Sleep(150 * time.Millisecond)
|
||||||
|
assert.True(t, batcher.ShouldFlush())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestClear(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
assert.Equal(t, 1, batcher.PendingBatchCount())
|
||||||
|
|
||||||
|
batcher.Clear()
|
||||||
|
assert.Equal(t, 0, batcher.QueueSize())
|
||||||
|
assert.Equal(t, 0, batcher.PendingBatchCount())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetQueuedOps(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
ops := []*GitOp{
|
||||||
|
{OpType: "commit", Message: "Commit 1"},
|
||||||
|
{OpType: "commit", Message: "Commit 2"},
|
||||||
|
{OpType: "commit", Message: "Commit 3"},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, op := range ops {
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
|
||||||
|
queued := batcher.GetQueuedOps()
|
||||||
|
assert.Equal(t, 3, len(queued))
|
||||||
|
assert.Equal(t, "Commit 1", queued[0].Message)
|
||||||
|
assert.Equal(t, "Commit 3", queued[2].Message)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCalculateNetworkSavings(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(3, 5*time.Second)
|
||||||
|
|
||||||
|
// Add 6 ops (will create 2 batches of 3 each)
|
||||||
|
for i := 0; i < 6; i++ {
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Mark both batches as completed
|
||||||
|
for i := 0; i < 2; i++ {
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
if batch != nil {
|
||||||
|
batcher.MarkBatchCompleted(batch.ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
savings := batcher.CalculateNetworkSavings()
|
||||||
|
// 2 batches of 3 each saves 4 round trips (3-1 + 3-1)
|
||||||
|
assert.Equal(t, 4, savings)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetBatchByID(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
batcher.Flush()
|
||||||
|
|
||||||
|
batch := batcher.GetPendingBatch()
|
||||||
|
retrieved := batcher.GetBatchByID(batch.ID)
|
||||||
|
|
||||||
|
assert.NotNil(t, retrieved)
|
||||||
|
assert.Equal(t, batch.ID, retrieved.ID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestGetBatchInfo(t *testing.T) {
|
||||||
|
batch := &GitBatch{
|
||||||
|
ID: "test-batch",
|
||||||
|
Status: "completed",
|
||||||
|
CreatedAt: time.Now(),
|
||||||
|
ExecutedAt: time.Now().Add(1 * time.Second),
|
||||||
|
}
|
||||||
|
|
||||||
|
info := batch.GetInfo()
|
||||||
|
assert.Equal(t, "test-batch", info["id"])
|
||||||
|
assert.Equal(t, "completed", info["status"])
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMultipleBatches(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(3, 5*time.Second)
|
||||||
|
|
||||||
|
// Create 3 batches
|
||||||
|
for batch := 0; batch < 3; batch++ {
|
||||||
|
for i := 0; i < 3; i++ {
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// All 3 batches should be pending
|
||||||
|
assert.Equal(t, 3, batcher.PendingBatchCount())
|
||||||
|
assert.Equal(t, 0, batcher.QueueSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestEnqueueNil(t *testing.T) {
|
||||||
|
batcher := NewGitBatcher(10, 5*time.Second)
|
||||||
|
|
||||||
|
// Enqueueing nil should not fail
|
||||||
|
batcher.Enqueue(nil)
|
||||||
|
assert.Equal(t, 0, batcher.QueueSize())
|
||||||
|
}
|
||||||
|
|
||||||
|
func BenchmarkEnqueue(b *testing.B) {
|
||||||
|
batcher := NewGitBatcher(1000, 10*time.Second)
|
||||||
|
|
||||||
|
for i := 0; i < b.N; i++ {
|
||||||
|
op := &GitOp{
|
||||||
|
OpType: "commit",
|
||||||
|
Branch: "main",
|
||||||
|
Message: "Commit",
|
||||||
|
}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func BenchmarkFlush(b *testing.B) {
|
||||||
|
batcher := NewGitBatcher(1000, 10*time.Second)
|
||||||
|
|
||||||
|
for i := 0; i < b.N; i++ {
|
||||||
|
op := &GitOp{OpType: "commit"}
|
||||||
|
batcher.Enqueue(op)
|
||||||
|
|
||||||
|
if (i + 1) % 100 == 0 {
|
||||||
|
batcher.Flush()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+1
-1
@@ -8,7 +8,7 @@
|
|||||||
| T2.2 | Parallel task dispatch: multiple T0.x tasks execute truly concurrently (not sequential) | [x] | `task/T2.2` | 9 tasks complete in ~1/9 total time (wall-clock speedup measured) |
|
| T2.2 | Parallel task dispatch: multiple T0.x tasks execute truly concurrently (not sequential) | [x] | `task/T2.2` | 9 tasks complete in ~1/9 total time (wall-clock speedup measured) |
|
||||||
| T2.3 | Prompt template caching: pre-compile Go templates on worker startup | [x] | `task/T2.3` | Template render latency < 100ms (vs parse+render each time) |
|
| T2.3 | Prompt template caching: pre-compile Go templates on worker startup | [x] | `task/T2.3` | Template render latency < 100ms (vs parse+render each time) |
|
||||||
| T2.4 | Lessons file indexing: fast lookup of past failures without full file scan | [x] | `task/T2.4` | Query lessons by task type → return in < 10ms for 1000s of entries |
|
| T2.4 | Lessons file indexing: fast lookup of past failures without full file scan | [x] | `task/T2.4` | Query lessons by task type → return in < 10ms for 1000s of entries |
|
||||||
| T2.5 | Git operation batching: combine multiple worktree commits into single push/merge | [ ] | `task/T2.5` | N tasks → 1 push (vs N pushes), measured via git ref-log |
|
| T2.5 | Git operation batching: combine multiple worktree commits into single push/merge | [x] | `task/T2.5` | N tasks → 1 push (vs N pushes), measured via git ref-log |
|
||||||
| T2.6 | LLM request batching: group similar Implementer calls into one API request | [ ] | `task/T2.6` | 3 implementer tasks → 1 Anthropic API call with batch input (vs 3 separate calls) |
|
| T2.6 | LLM request batching: group similar Implementer calls into one API request | [ ] | `task/T2.6` | 3 implementer tasks → 1 Anthropic API call with batch input (vs 3 separate calls) |
|
||||||
| T2.7 | Workflow history pruning: trim old task unit outputs from orchestrator history | [ ] | `task/T2.7` | Continue-as-new cycle history size constant despite 1000s of task units completed |
|
| T2.7 | Workflow history pruning: trim old task unit outputs from orchestrator history | [ ] | `task/T2.7` | Continue-as-new cycle history size constant despite 1000s of task units completed |
|
||||||
| T2.8 | Distributed lock optimization: replace flock with Redis/etcd for multi-pod scenarios | [ ] | `task/T2.8` | 5 concurrent orchestrators on different pods share FS safely via distributed lock |
|
| T2.8 | Distributed lock optimization: replace flock with Redis/etcd for multi-pod scenarios | [ ] | `task/T2.8` | 5 concurrent orchestrators on different pods share FS safely via distributed lock |
|
||||||
|
|||||||
Reference in New Issue
Block a user