[$50 BOUNTY] [Go] Make analytics collector Start idempotent

docs/OPERATIONS.md

@@ -310,3 +310,16 @@ Audit logs are retained for 365 days and include:
 2. Update Kubernetes secret: `kubectl create secret tls tot-tls --cert=new.crt --key=new.key -n tent-production --dry-run=client -o yaml | kubectl apply -f -`
 3. Restart services: `kubectl rollout restart deployment -n tent-production`
 4. Verify new certificate: `openssl s_client -connect api.example.com:443 -servername api.example.com`
+
+## Analytics Collector Idempotent Start
+
+The `market/analytics/collector.go` `Collector.Start()` method is now idempotent.
+Calling `Start()` more than once returns `ErrAlreadyStarted` without spawning
+duplicate flush goroutines. After `Stop()` or context cancellation, the collector
+can be started again.
+
+### Key Changes
+
+- `Start(ctx)` now returns `error` — nil on first call, `ErrAlreadyStarted` on repeat
+- `IsStarted()` reports whether the flush loop is currently active
+- After `Stop()` or context cancellation, `IsStarted()` returns false and `Start()` may be called again

market/analytics/collector.go

@@ -16,6 +16,7 @@ import (
 	"context"
 	"encoding/csv"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
 	"math/rand"
@@ -25,6 +26,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -299,6 +301,7 @@ type Collector struct {
 	flushInterval time.Duration
 	maxBacklog    int
 	stopCh        chan struct{}
+	started       atomic.Bool
 	flushed       int64
 	errors        int64
 	dropped       int64
@@ -455,13 +458,26 @@ func (c *Collector) RecordHistogram(name string, value float64, tags ...MetricTa
 	})
 }
 
+// ErrAlreadyStarted is returned by Start() when the collector is already running.
+var ErrAlreadyStarted = errors.New("analytics: collector already started")
+
+// IsStarted reports whether the collector's flush loop is currently active.
+func (c *Collector) IsStarted() bool {
+	return c.started.Load()
+}
+
 // Start begins the background flush loop. It spawns a goroutine that
 // periodically flushes collected metrics to the backend. The flush
 // loop will stop when the context is cancelled or Stop() is called.
-// NOTE: Calling Start() multiple times will spawn multiple flush
-// goroutines, causing duplicate flushes. This is a known issue.
-// TODO: Make Start() idempotent.
-func (c *Collector) Start(ctx context.Context) {
+//
+// Start is idempotent: calling it more than once returns ErrAlreadyStarted
+// without spawning additional goroutines. Callers can check IsStarted()
+// before calling Start if they need to distinguish first-start from
+// repeated-start in application logic.
+func (c *Collector) Start(ctx context.Context) error {
+	if !c.started.CompareAndSwap(false, true) {
+		return ErrAlreadyStarted
+	}
 	go func() {
 		// Tick immediately to flush any bootstrapped metrics
 		c.flush(ctx)
@@ -472,14 +488,17 @@ func (c *Collector) Start(ctx context.Context) {
 			case <-ctx.Done():
 				// Final flush before exiting
 				c.flush(context.Background())
+				c.started.Store(false)
 				return
 			case <-c.stopCh:
+				c.started.Store(false)
 				return
 			case <-ticker.C:
 				c.flush(ctx)
 			}
 		}
 	}()
+	return nil
 }
 
 // Stop signals the flush loop to stop. It does NOT perform a final flush.

market/analytics/collector_idempotent_test.go

@@ -0,0 +1,156 @@
+package analytics
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestCollector_SingleStart(t *testing.T) {
+	c := NewCollector()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	err := c.Start(ctx)
+	if err != nil {
+		t.Fatalf("first Start() should succeed, got: %v", err)
+	}
+	if !c.IsStarted() {
+		t.Fatal("IsStarted() should be true after Start()")
+	}
+}
+
+func TestCollector_RepeatedStart(t *testing.T) {
+	c := NewCollector()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// First call should succeed
+	if err := c.Start(ctx); err != nil {
+		t.Fatalf("first Start() should succeed, got: %v", err)
+	}
+
+	// Second call should return ErrAlreadyStarted
+	if err := c.Start(ctx); !errors.Is(err, ErrAlreadyStarted) {
+		t.Fatalf("second Start() should return ErrAlreadyStarted, got: %v", err)
+	}
+
+	// Third call should also return ErrAlreadyStarted
+	if err := c.Start(ctx); !errors.Is(err, ErrAlreadyStarted) {
+		t.Fatalf("third Start() should return ErrAlreadyStarted, got: %v", err)
+	}
+}
+
+func TestCollector_StopAfterRepeatedStart(t *testing.T) {
+	c := NewCollector()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Start the collector
+	if err := c.Start(ctx); err != nil {
+		t.Fatalf("Start() should succeed, got: %v", err)
+	}
+
+	// Try to start again (should fail)
+	if err := c.Start(ctx); !errors.Is(err, ErrAlreadyStarted) {
+		t.Fatalf("second Start() should return ErrAlreadyStarted, got: %v", err)
+	}
+
+	// Stop should work
+	c.Stop()
+	time.Sleep(50 * time.Millisecond) // give goroutine time to exit
+
+	// After stop, IsStarted should be false
+	if c.IsStarted() {
+		t.Fatal("IsStarted() should be false after Stop()")
+	}
+}
+
+func TestCollector_StartAfterStop(t *testing.T) {
+	c := NewCollector()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Start and then stop
+	if err := c.Start(ctx); err != nil {
+		t.Fatalf("first Start() should succeed, got: %v", err)
+	}
+	c.Stop()
+	time.Sleep(50 * time.Millisecond)
+
+	// After stop, we should be able to start again
+	if err := c.Start(ctx); err != nil {
+		t.Fatalf("Start() after Stop() should succeed, got: %v", err)
+	}
+	if !c.IsStarted() {
+		t.Fatal("IsStarted() should be true after restart")
+	}
+}
+
+func TestCollector_StartContextCancel(t *testing.T) {
+	c := NewCollector()
+	ctx, cancel := context.WithCancel(context.Background())
+
+	if err := c.Start(ctx); err != nil {
+		t.Fatalf("Start() should succeed, got: %v", err)
+	}
+
+	// Cancel the context
+	cancel()
+	time.Sleep(50 * time.Millisecond)
+
+	// After context cancellation, IsStarted should be false
+	if c.IsStarted() {
+		t.Fatal("IsStarted() should be false after context cancellation")
+	}
+}
+
+func TestCollector_ConcurrentStart(t *testing.T) {
+	c := NewCollector()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+	errs := make([]error, 10)
+
+	// Try to start from 10 goroutines simultaneously
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+			errs[idx] = c.Start(ctx)
+		}(i)
+	}
+	wg.Wait()
+
+	// Exactly one goroutine should have succeeded (nil error)
+	successCount := 0
+	for _, err := range errs {
+		if err == nil {
+			successCount++
+		}
+	}
+	if successCount != 1 {
+		t.Fatalf("expected exactly 1 successful Start(), got %d", successCount)
+	}
+
+	// The rest should have gotten ErrAlreadyStarted
+	errAlreadyCount := 0
+	for _, err := range errs {
+		if errors.Is(err, ErrAlreadyStarted) {
+			errAlreadyCount++
+		}
+	}
+	if errAlreadyCount != 9 {
+		t.Fatalf("expected 9 ErrAlreadyStarted errors, got %d", errAlreadyCount)
+	}
+}
+
+func TestCollector_IsStartedBeforeStart(t *testing.T) {
+	c := NewCollector()
+	if c.IsStarted() {
+		t.Fatal("IsStarted() should be false before Start() is called")
+	}
+}