seg: SamplingFactor>1 produced only 1 sample (erigontech#21806)

## Problem

`Compressor.AddWord` decided when a superstring window was full from
`len(c.superstring)`. But skipped (non-sampled) windows never append to
that buffer, so once `SamplingFactor>1` reached the first skipped window
the buffer stopped growing, the overflow branch never fired again, the
window counter froze, and **the pattern dictionary was built from the
first `superstringLimit` (16MB) of the file only.**

This affects every `SamplingFactor>1` user: `seg.DefaultCfg` and
`BlockCompressCfg` both use `SamplingFactor=4`, so tx/bodies/headers
segments have been built from first-16MB dictionaries.

## Fix

Track scanned bytes per window in `scannedBytes`, advanced on **every**
word regardless of sampling, and use it (not `len(c.superstring)`) for
the overflow check. Window boundaries now advance through skipped
windows, so `SamplingFactor` honestly samples every Nth window across
the whole file.

While here, the windowing logic moved into a small `advanceScan` helper
so the byte accounting can't drift from the rollover it feeds, and the
send condition is now `len(superstring) > 0` (non-empty ⟺ sampled): this
stops pushing empty buffers to the workers and only fetches a fresh pool
buffer when one is actually handed off.

## Scope / impact

- **Affected:** `seg.DefaultCfg`, `BlockCompressCfg`
(`SamplingFactor=4`) — tx/bodies/headers segments now build from a true
25% whole-file sample (slower build, smaller output; offset by erigontech#21625's
matcher speedups).
- **Not affected:** `DomainCompressCfg` / `HistoryCompressCfg`
(`SamplingFactor=1`) — for `SamplingFactor=1` the new accounting is
provably identical to the old, so domain/history snapshots stay
**byte-identical** (verified by the unchanged checksum tests).

Addresses the core bug in erigontech#21628. Per-config `SamplingFactor` re-tuning
(the issue's other open thread) is intentionally left out of this PR.

## Testing

- New `TestCompressSamplingCoversWholeFile` pins the invariant "the
number of windows a file splits into must not depend on
`SamplingFactor`" — fails on the old code (SF=1 → 90 windows, SF=4 →
stuck at 1), passes after the fix.
- Full `db/seg` suite passes under `-race`; existing checksum-asserting
tests unchanged (single-window output byte-identical).

Author: AskAlexSharov

db/seg/compress.go

@@ -123,6 +123,8 @@ type Compressor struct {
 	// this is needed for using ordinary (one string) suffix sorting algorithm instead of a generalised (many superstrings) suffix
 	// sorting algorithm
 	superstring       []byte
+	scannedBytes      int
+	superstringLimit  int
 	wordsCount        uint64
 	superstringCount  uint64
 	uncompressedBytes int
@@ -187,6 +189,7 @@ func NewCompressor(ctx context.Context, logPrefix, outputFile, tmpDir string, cf
 		wg:               wg,
 		logger:           logger,
 		version:          FileCompressionFormatV1,
+		superstringLimit: superstringLimit,
 	}
 
 	if cfg.ValuesOnCompressedPage > 0 {
@@ -267,16 +270,8 @@ func (c *Compressor) AddWord(word []byte) error {
 		}
 	}
 
-	l := 2*len(word) + 2
-	if len(c.superstring)+l > superstringLimit {
-		if c.superstringCount%c.SamplingFactor == 0 {
-			c.superstrings <- c.superstring
-		}
-		c.superstringCount++
-		c.superstring = superStringsPool.Get().([]byte)[:0]
-	}
-
-	if c.superstringCount%c.SamplingFactor == 0 {
+	c.advanceScan(2*len(word) + 2)
+	if c.sampledSuperstring() {
 		for _, a := range word {
 			c.superstring = append(c.superstring, 1, a)
 		}
@@ -287,6 +282,25 @@ func (c *Compressor) AddWord(word []byte) error {
 	return c.uncompressedFile.Append(word)
 }
 
+func (c *Compressor) sampledSuperstring() bool { return c.superstringCount%c.SamplingFactor == 0 }
+
+// advanceScan accounts for the next word (encoded size l) and cuts a new superstring when the
+// current one would overflow. Bytes are counted for every word so boundaries don't depend on
+// sampling; a superstring fills only when sampled, so only a non-empty buffer is sent and replaced.
+func (c *Compressor) advanceScan(l int) {
+	if c.scannedBytes+l <= c.superstringLimit {
+		c.scannedBytes += l
+		return
+	}
+
+	if len(c.superstring) > 0 {
+		c.superstrings <- c.superstring
+		c.superstring = superStringsPool.Get().([]byte)[:0]
+	}
+	c.superstringCount++
+	c.scannedBytes = 0
+}
+
 func (c *Compressor) AddUncompressedWord(word []byte) error {
 	c.wordsCount++
 	if c.wordsCount%1024 == 0 {

db/seg/compress_test.go

@@ -387,3 +387,40 @@ func TestCompressNoWordPatterns(t *testing.T) {
 		t.Errorf("fast-path output differs from main, checksum=%d", cs)
 	}
 }
+
+// The number of superstring windows a file is split into is a property of the data, so it
+// must not depend on SamplingFactor — sampling only selects which windows feed the dict.
+func TestCompressSamplingCoversWholeFile(t *testing.T) {
+	const word = "mykeyabc"
+	const nWords = 5000
+
+	windowCount := func(samplingFactor uint64) uint64 {
+		logger := log.New()
+		tmpDir := t.TempDir()
+		file := filepath.Join(tmpDir, "compressed")
+		cfg := DefaultCfg
+		cfg.MinPatternScore = 1
+		cfg.SamplingFactor = samplingFactor
+		c, err := NewCompressor(t.Context(), t.Name(), file, tmpDir, cfg, log.LvlDebug, logger)
+		require.NoError(t, err)
+		defer c.Close()
+		c.superstringLimit = 1000
+
+		for i := 0; i < nWords; i++ {
+			require.NoError(t, c.AddWord([]byte(word)))
+		}
+		count := c.superstringCount
+		require.NoError(t, c.Compress())
+
+		d, err := NewDecompressor(file)
+		require.NoError(t, err)
+		defer d.Close()
+		require.EqualValues(t, nWords, d.Count())
+		return count
+	}
+
+	full := windowCount(1)
+	require.Greater(t, full, uint64(1))
+	require.Equal(t, full, windowCount(4))
+	require.Equal(t, full, windowCount(8))
+}