fix: expire boundary double-count and label sort order in counter chains

cmd/capture-fixtures/main.go

@@ -0,0 +1,179 @@
+// Command capture-fixtures fetches Cloudflare observations for a time range
+// and writes them as trimmed JSON fixtures for use in spike_test.go.
+//
+// Usage:
+//
+//	go run ./cmd/capture-fixtures \
+//	  -zone 43c0ec06d9a970d6707dee37374c1b13 \
+//	  -start 2026-06-26T14:00:00Z \
+//	  -end   2026-06-26T14:22:00Z \
+//	  -lookback 25m \
+//	  -prefix spike_0626 \
+//	  -metric cloudflare_zone_requests_total \
+//	  -outdir converge/testdata
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	cfzones "github.com/cloudflare/cloudflare-go/v4/zones"
+	"github.com/lablabs/cloudflare-exporter/cfetch"
+	"github.com/lablabs/cloudflare-exporter/cfgql"
+	"github.com/lablabs/cloudflare-exporter/converge"
+)
+
+type gqlClient struct{ token string }
+
+type gqlResponse struct {
+	Data   any        `json:"data"`
+	Errors []gqlError `json:"errors"`
+}
+type gqlError struct {
+	Message string `json:"message"`
+}
+
+func (c *gqlClient) RunGQL(ctx context.Context, req *cfgql.GQLRequest, dest any) error {
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	body, _ := json.Marshal(req)
+	httpReq, _ := http.NewRequestWithContext(ctx, http.MethodPost,
+		"https://api.cloudflare.com/client/v4/graphql/",
+		bytes.NewReader(body))
+	httpReq.Header.Set("Accept", "application/json; charset=utf-8")
+	httpReq.Header.Set("Authorization", "Bearer "+c.token)
+
+	resp, err := http.DefaultClient.Do(httpReq)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	var gResp gqlResponse
+	gResp.Data = dest
+	if err := json.NewDecoder(resp.Body).Decode(&gResp); err != nil {
+		return err
+	}
+	for _, e := range gResp.Errors {
+		return fmt.Errorf("graphql: %s", e.Message)
+	}
+	return nil
+}
+
+type recordedFetch struct {
+	Seq          int                   `json:"seq"`
+	Start        time.Time             `json:"start"`
+	End          time.Time             `json:"end"`
+	Observations []recordedObservation `json:"observations"`
+}
+
+type recordedObservation struct {
+	Name   string            `json:"name"`
+	Labels map[string]string `json:"labels"`
+	Value  uint64            `json:"value"`
+	Bucket time.Time         `json:"bucket"`
+}
+
+func main() {
+	var (
+		zoneID = flag.String("zone", "", "Cloudflare zone ID")
+		startS = flag.String("start", "", "start time (RFC3339)")
+		endS   = flag.String("end", "", "end time (RFC3339)")
+		lookb  = flag.Duration("lookback", 25*time.Minute, "lookback duration for the live fetch")
+		chunk  = flag.Duration("chunk", 10*time.Minute, "backfill chunk size")
+		prefix = flag.String("prefix", "spike", "filename prefix")
+		metric = flag.String("metric", "cloudflare_zone_requests_total", "metric to keep (empty = all)")
+		outdir = flag.String("outdir", "converge/testdata", "output directory")
+	)
+	flag.Parse()
+
+	token := os.Getenv("CF_API_TOKEN")
+	if token == "" {
+		log.Fatal("CF_API_TOKEN required")
+	}
+	if *zoneID == "" || *startS == "" || *endS == "" {
+		log.Fatal("-zone, -start, -end are required")
+	}
+
+	start, err := time.Parse(time.RFC3339, *startS)
+	if err != nil {
+		log.Fatalf("bad -start: %v", err)
+	}
+	end, err := time.Parse(time.RFC3339, *endS)
+	if err != nil {
+		log.Fatalf("bad -end: %v", err)
+	}
+
+	client := &gqlClient{token: token}
+	zones := []cfzones.Zone{{ID: *zoneID, Name: "captured"}}
+	fetcher := cfetch.New(client, zones, nil)
+	ctx := context.Background()
+
+	os.MkdirAll(*outdir, 0755)
+
+	seq := 0
+
+	// Live fetch: [end - lookback, end]
+	liveStart := end.Add(-*lookb)
+	writeFetch(ctx, fetcher, liveStart, end, seq, *prefix, *metric, *outdir)
+	seq++
+
+	// Backfill: [start, end - lookback] in chunks
+	cursor := start
+	limit := end.Add(-*lookb)
+	for cursor.Before(limit) {
+		chunkEnd := cursor.Add(*chunk)
+		if chunkEnd.After(limit) {
+			chunkEnd = limit
+		}
+		writeFetch(ctx, fetcher, cursor, chunkEnd, seq, *prefix, *metric, *outdir)
+		seq++
+		cursor = chunkEnd
+	}
+
+	log.Printf("wrote %d fixture files to %s", seq, *outdir)
+}
+
+func writeFetch(ctx context.Context, f converge.Fetcher, start, end time.Time, seq int, prefix, metric, outdir string) {
+	obs, err := f.Fetch(ctx, start, end)
+	if err != nil {
+		log.Printf("WARN: fetch [%s, %s] failed: %v", start, end, err)
+		return
+	}
+
+	rec := recordedFetch{
+		Seq:   seq,
+		Start: start,
+		End:   end,
+	}
+	for _, o := range obs {
+		if metric != "" && o.Key.Name != metric {
+			continue
+		}
+		labels := make(map[string]string, len(o.Key.Labels))
+		for _, l := range o.Key.Labels {
+			labels[l.Name] = l.Value
+		}
+		rec.Observations = append(rec.Observations, recordedObservation{
+			Name:   o.Key.Name,
+			Labels: labels,
+			Value:  o.Value,
+			Bucket: o.Bucket,
+		})
+	}
+
+	path := fmt.Sprintf("%s/%s_%03d.json", outdir, prefix, seq)
+	data, _ := json.MarshalIndent(rec, "", "  ")
+	os.WriteFile(path, data, 0600)
+	log.Printf("%s: %d obs (filtered from %d) [%s, %s]",
+		path, len(rec.Observations), len(obs),
+		start.Format(time.RFC3339), end.Format(time.RFC3339))
+}

converge.go

@@ -15,7 +15,6 @@ import (
 
 const (
 	argConvergeThreshold       = "converge_threshold"
-	argConvergeWindowTTL       = "converge_window_ttl"
 	argConvergePollInterval    = "converge_poll_interval"
 	argConvergeLookback        = "converge_lookback"
 	argConvergeMaxBackfill     = "converge_max_backfill"
@@ -123,7 +122,6 @@ func cfetchdnsEnabledSet() map[string]bool {
 func convergeConfig() converge.Config {
 	cfg := converge.DefaultConfig()
 	cfg.Threshold = viper.GetInt(argConvergeThreshold)
-	cfg.WindowTTL = viper.GetDuration(argConvergeWindowTTL)
 	cfg.PollInterval = viper.GetDuration(argConvergePollInterval)
 	cfg.Lookback = viper.GetDuration(argConvergeLookback)
 	cfg.MaxBackfill = viper.GetDuration(argConvergeMaxBackfill)
@@ -142,6 +140,20 @@ func setupConvergerWithFetcher(ctx context.Context, component string, fetcher co
 		return nil, err
 	}
 	cfg := convergeConfig()
+	cfg.OnTick = func(ts converge.TickStats) {
+		convergeIngestSamples.WithLabelValues(component).Add(float64(ts.IngestSamples))
+		convergePostStabilizeUpdates.WithLabelValues(component).Add(float64(ts.PostStabilizeUpdates))
+		convergeExpireFlushes.WithLabelValues(component).Add(float64(ts.ExpireFlushes))
+		convergeExpireSamples.WithLabelValues(component).Add(float64(ts.ExpireSamples))
+		convergeLiveFetchObservations.WithLabelValues(component).Add(float64(ts.LiveObservations))
+		convergeBackfillFetchObservations.WithLabelValues(component).Add(float64(ts.BackfillObservations))
+		convergeSnapshotSamples.WithLabelValues(component).Add(float64(ts.SnapshotSamples))
+		convergeOpenWindows.WithLabelValues(component).Set(float64(ts.OpenWindows))
+		convergeTrackerCount.WithLabelValues(component).Set(float64(ts.TrackerCount))
+		convergeGaugeDownRevisions.WithLabelValues(component).Add(float64(ts.GaugeDownRevisions))
+		convergeCounterRegressions.WithLabelValues(component).Add(float64(ts.CounterRegressions))
+		convergePushErrors.WithLabelValues(component).Add(float64(ts.PushErrors))
+	}
 	return func(ctx context.Context) error {
 		return converge.Run(
 			converge.ContextWithLogger(ctx, log.WithField("component", component)),

converge/README.md

@@ -50,9 +50,8 @@ eng := converge.NewEngine(cfg)
 // Feed observations. Returns samples for any tracker that just stabilized.
 samples := eng.Ingest(observations)
 
-// Check TTLs. Returns forced-push samples from expiring windows.
-// Removes closed windows.
-samples = eng.Expire(time.Now())
+// Expire windows older than Lookback. Returns forced-push samples.
+samples = eng.Expire(time.Now(), true)
 
 // Graceful shutdown: flush best-known values for all open windows.
 samples = eng.Flush()
@@ -141,9 +140,8 @@ Ingest/Expire/Flush yourself.
 ```go
 converge.Config{
     Threshold:            3,              // identical observations to stabilize
-    WindowTTL:            15 * time.Minute, // window lifespan
     PollInterval:         30 * time.Second, // live lane tick
-    Lookback:             10 * time.Minute, // live query range
+    Lookback:             10 * time.Minute, // live query range and window TTL
     MaxBackfill:          2 * time.Hour,    // startup backfill cap
     BackfillChunk:        10 * time.Minute, // time range per backfill call
     BackfillCallsPerTick: 1,               // API budget for backfill per tick
@@ -157,6 +155,31 @@ Eager:        Threshold=1  -- push on first sight, correct later
 Conservative: Threshold=3  -- wait for convergence, push once
 ```
 
+## Prometheus metrics
+
+The root package registers per-tick counters and gauges on `/metrics`,
+labeled by `component` (`converge` or `converge-dns`). These help tune
+Threshold, Lookback, and MaxBackfill by making the tradeoffs observable.
+
+| Metric | Type | What it tells you |
+|--------|------|-------------------|
+| `converge_ingest_samples_total` | counter | Tracker stabilizations. Responds to threshold tuning. |
+| `converge_post_stabilize_updates_total` | counter | Values that changed after push. High = threshold too low. |
+| `converge_expire_flushes_total` | counter | Windows force-flushed without stabilizing. High = threshold too high. |
+| `converge_expire_samples_total` | counter | Samples from TTL expiry (corrections on eviction). |
+| `converge_live_fetch_observations_total` | counter | Raw observation volume from CF live fetches. |
+| `converge_backfill_fetch_observations_total` | counter | Raw observation volume from backfill. |
+| `converge_snapshot_samples_total` | counter | Samples in the post-backfill snapshot. |
+| `converge_open_windows` | gauge | Current active windows. |
+| `converge_tracker_count` | gauge | Current active trackers. |
+| `converge_push_errors_total` | counter | Failed pushes to sink. |
+
+Tuning signals:
+
+- `post_stabilize_updates` climbing: threshold is too low, pushing before CF data settles.
+- `expire_flushes` climbing: threshold is too high, values never converge within the lookback window.
+- `open_windows` growing without bound: expire boundary issue or backfill stuck.
+
 ## Integration
 
 The engine defines two interfaces:
@@ -243,6 +266,72 @@ the VM query endpoint, derived from the write endpoint.
 
 Reference: [VictoriaMetrics Storage, Retention, Merging, and Deduplication](https://victoriametrics.com/blog/vmstorage-retention-merging-deduplication/)
 
+## Why gauge-to-counter conversion is fundamentally hard
+
+Cloudflare's API returns **gauges**: independent per-minute counts. Each
+bucket stands alone. "742 requests in the 14:10 minute" has no relationship
+to any other bucket.
+
+Our dashboards expect **counters**: a monotonically increasing value where
+`rate()` computes the derivative. Counter at 14:10 = sum of all gauges from
+the beginning of time.
+
+The conversion (prefix sum) creates three problems that don't exist with
+raw gauges:
+
+**1. State that outlives a process.** A gauge is stateless. A counter
+carries accumulated history. When the exporter restarts, the accumulated
+base is lost. The counter resets to near zero, and `rate()` interprets the
+drop as either a reset (handled) or a massive negative spike (not handled
+by VM's "highest value wins" dedup). Gauges don't have this problem because
+each value is self-contained.
+
+**2. Ordering dependencies between writes.** With gauges, you can push
+bucket 14:10 before 14:05 and nothing goes wrong. With counters, the value
+at 14:10 depends on every bucket before it. If a backfill chunk arrives and
+inserts earlier data, all later counter values shift (the cascade). You must
+either serialize all writes or suppress intermediate states. Gauges have no
+such dependency.
+
+**3. Corrections amplify instead of replacing.** When CF revises a gauge
+(700 to 742), it's a simple overwrite. When we convert to a counter, that
++42 correction cascades forward: every subsequent counter shifts by +42 and
+gets re-pushed. Combined with VM's "highest value wins" dedup, the old
+higher values from before the correction persist forever. Gauges just
+overwrite in place.
+
+### The core tension
+
+We're trying to maintain a **cumulative running total** from a source that
+provides **independent snapshots**, push it to a store that **doesn't
+support last-write-wins**, and do it across **process restarts** without
+losing the accumulated state.
+
+Each of those properties in isolation is manageable. Together they create a
+combinatorial surface of edge cases: restart + backfill + correction + dedup
+= spike artifacts that are permanent in VM.
+
+### What we've built to manage it
+
+- Suppress + snapshot (don't push during backfill, push final state once)
+- Chain base seeding (read back last value from VM on restart)
+- Expire boundary fix (prevent double-counting at the lookback edge)
+- Label sort normalization (prevent duplicate chains)
+- Prometheus observability counters (make tuning data-driven)
+- Integration tests + fixture capture (reproduce and regression-test spikes)
+
+Each fix addresses a real incident, but the root cause is the same:
+counters have global state, gauges don't.
+
+### The alternative
+
+Push gauges to VM and let the query layer compute `sum(increase(...))`
+instead of `rate()`. This eliminates all the complexity above: no chain
+state, no base seeding, no cascade, no dedup sensitivity, no restart
+artifacts. The tradeoff is that existing dashboards using `rate()` would
+need to change, and `increase()` over gauges is slightly less ergonomic
+than `rate()` over counters.
+
 ## File layout
 
 ```

converge/counter_chain.go

@@ -22,13 +22,15 @@ type counterEmission struct {
 // and the chain reports which values need re-pushing.
 //
 // This is a prefix sum array with point-update support. For the small number
-// of active buckets (bounded by WindowTTL / bucket_interval, roughly 15), a
+// of active buckets (bounded by Lookback / bucket_interval, roughly 15), a
 // sorted slice with linear recomputation is optimal. A Fenwick tree (Binary
 // Indexed Tree) solves the same problem class in O(log n) per operation but
 // adds complexity that isn't justified at this scale.
 type counterChain struct {
-	base    uint64       // accumulated sum of evicted (expired) buckets
-	entries []chainEntry // sorted ascending by bucket time
+	base               uint64       // accumulated sum of evicted (expired) buckets
+	entries            []chainEntry // sorted ascending by bucket time
+	gaugeDownRevisions uint64       // CF revised a gauge downward
+	counterRegressions uint64       // computed counter was lower than previous emission
 }
 
 type chainEntry struct {
@@ -52,6 +54,9 @@ func (c *counterChain) Set(bucket time.Time, gauge uint64) []counterEmission {
 		if c.entries[idx].gauge == gauge {
 			return nil
 		}
+		if gauge < c.entries[idx].gauge {
+			c.gaugeDownRevisions++
+		}
 		c.entries[idx].gauge = gauge
 	} else {
 		// New entry: insert at sorted position.
@@ -73,6 +78,9 @@ func (c *counterChain) Set(bucket time.Time, gauge uint64) []counterEmission {
 		newCounter := prev + c.entries[i].gauge
 
 		if newCounter != c.entries[i].counter || i == idx {
+			if c.entries[i].counter > 0 && newCounter < c.entries[i].counter {
+				c.counterRegressions++
+			}
 			c.entries[i].counter = newCounter
 			emissions = append(emissions, counterEmission{
 				Bucket:  c.entries[i].bucket,