storage: wire new wavesdb engine capabilities into waveSpan

cmd/wavespan-node/main.go

@@ -218,9 +218,16 @@ func run() error {
 
 	store, err := storage.OpenWavesdbWith(cfg.Storage.Path, engineOptions(tun))
 	if err != nil {
+		if errors.Is(err, storage.ErrLocked) {
+			// Another process (a still-running node, or a stale one that did not release the lock)
+			// holds the data dir. Two live processes on one dir would corrupt the store; refuse to start.
+			logger.Error("cannot start: data directory is already in use by another process", "path", cfg.Storage.Path)
+		}
 		return fmt.Errorf("open storage: %w", err)
 	}
 	defer func() { _ = store.Close() }()
+	// Per-CF engine counters (read amplification, bloom effectiveness, physical size) → Prometheus.
+	metrics.Registry.MustRegister(observability.NewEngineStatsCollector(store))
 	storageUUID, err := storage.EnsureStorageUUID(store)
 	if err != nil {
 		return fmt.Errorf("storage uuid: %w", err)
@@ -443,6 +450,16 @@ func run() error {
 		ResumeFreePct:   tun.Get("disk.resumeFreePct").Float(),
 		CriticalFreePct: tun.Get("disk.criticalFreePct").Float(),
 		CheckInterval:   tun.Get("disk.checkInterval").Duration(),
+		// Fail-stop poisoning (design/36): if the engine rejects writes after a durability failure
+		// (fsync/flush/manifest), the disk gate sheds all writes and we deliberately shut the node down.
+		// The orchestrator restarts it; wavesdb re-runs recovery on reopen and clears the poison (or
+		// re-poisons if the fault persists — a clean fail-stop rather than the collections-raft ENOSPC
+		// panic/crash-loop this admission control exists to prevent).
+		Poisoner: store,
+		OnPoison: func(err error) {
+			logger.Error("storage engine poisoned — fail-stopping node; restart re-runs recovery", "err", err, "path", cfg.Storage.Path)
+			stop() // cancel the root context → existing graceful-shutdown path → process exits
+		},
 	}, diskMetrics)
 	diskMon.Start(ctx)
 	defer diskMon.Stop()
@@ -1276,6 +1293,7 @@ func engineOptions(t *tunables.Registry) storage.EngineOptions {
 		MinLevels:            g("minLevels").Int(),
 		KlogValueThreshold:   g("klogValueThreshold").Int(),
 		Compression:          g("compression").String(),
+		CompressionPerLevel:  splitCSV(g("compressionPerLevel").String()),
 		EnableBloomFilter:    g("enableBloomFilter").Bool(),
 		BloomFPR:             g("bloomFpr").Float(),
 		EnableBlockIndex:     g("enableBlockIndex").Bool(),
@@ -1292,6 +1310,22 @@ func engineOptions(t *tunables.Registry) storage.EngineOptions {
 	}
 }
 
+// splitCSV parses a comma-separated tunable (e.g. the per-level compression list) into trimmed,
+// non-empty tokens. An empty or whitespace-only string yields nil (the engine's "uniform" default).
+func splitCSV(s string) []string {
+	if strings.TrimSpace(s) == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	out := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if v := strings.TrimSpace(p); v != "" {
+			out = append(out, v)
+		}
+	}
+	return out
+}
+
 // applyRuntimeTunables applies the runtime.* tunables to the Go runtime and registers OnApply hooks
 // so a Hot change (later via gossip) takes effect live.
 func applyRuntimeTunables(t *tunables.Registry) {

config/reference.yaml

@@ -69,6 +69,16 @@ tunables:
       # enable zstd/snappy to trade CPU for space.
       compression: none
 
+      # [string, static] env: WAVESPAN_TUNABLE_STORAGE_ENGINE_COMPRESSION_PER_LEVEL
+      # Optional per-LSM-level codec override as a comma list, e.g. "none,none,zstd" (L0/L1
+      # uncompressed, L2+ zstd); the last entry repeats for deeper levels. Empty = use
+      # compression for every level.
+      # why: Leaves hot, recently-written L0/L1 data uncompressed for lowest write/read CPU
+      # while compressing cold, rarely-read deep levels — most of the space win at a
+      # fraction of the CPU cost of uniform compression. Blocks are self-describing, so this
+      # only affects newly written SSTables and never breaks reads of existing ones.
+      compressionPerLevel: ""
+
       # [bool, static] env: WAVESPAN_TUNABLE_STORAGE_ENGINE_ENABLE_BLOOM_FILTER
       # Build a per-SSTable bloom filter to skip non-matching files on point reads.
       # why: true: a small filter dramatically cuts disk reads for absent keys, which dominate

internal/backup/logical_restore.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"time"
 
 	"github.com/yannick/wavespan/internal/collections"
 	"github.com/yannick/wavespan/internal/storage"
@@ -139,6 +140,44 @@ func restoreCFObject(dst storage.LocalStore, store ObjectStore, objKey string, c
 
 	reshard := cf == storage.CFReplData && ri.CollectionsDataShards > 0
 
+	// Fast path: bulk-ingest directly into L0, bypassing the WAL/memtable. The export scans each CF in
+	// ascending key order, so the object stream is strictly ascending; skipping the identity key
+	// preserves that. Re-shard is EXCLUDED because rewriting the 8-byte shard prefix reorders keys, which
+	// the ingester rejects — it must use the batch path. Falls back to Batch when dst lacks the
+	// capability (e.g. MemStore in tests).
+	if ing, ok := dst.(storage.SortedIngester); ok && !reshard {
+		var decoded int64
+		if _, err := ing.IngestSorted(cf, func(emit func(key, value []byte, ttl time.Duration, tombstone bool) error) error {
+			for {
+				key, err := readBytes(br)
+				if err == io.EOF {
+					return nil
+				}
+				if err != nil {
+					return err
+				}
+				val, err := readBytes(br)
+				if err != nil {
+					return err
+				}
+				decoded++
+				// Preserve the target's own node identity (as in the batch path).
+				if cf == storage.CFSys && string(key) == storageIdentityKey {
+					continue
+				}
+				if err := emit(key, val, 0, false); err != nil {
+					return err
+				}
+			}
+		}); err != nil {
+			return err
+		}
+		if decoded != wantEntries {
+			return fmt.Errorf("backup: CF %q entry-count mismatch: manifest records %d, object decoded %d (truncated or corrupt)", cf.Name(), wantEntries, decoded)
+		}
+		return nil
+	}
+
 	var ops []storage.StoreOp
 	flush := func() error {
 		if len(ops) == 0 {

internal/health/diskmonitor.go

@@ -69,6 +69,16 @@ type Config struct {
 	// CheckInterval is how often the volume is polled. Default 5s.
 	CheckInterval time.Duration
 
+	// Poisoner, when non-nil, is polled each sample for an engine fail-stop (durability failure). On
+	// the first observed poison the monitor sheds all writes (UnderPressure becomes true), sets the
+	// poisoned metric, and fires OnPoison once. Poison is terminal — recovery is a process restart —
+	// so the flag never clears. nil disables the probe (e.g. the in-memory store).
+	Poisoner Poisoner
+	// OnPoison, when non-nil, is invoked exactly once the first time poison is observed. The node wires
+	// it to a loud log + graceful shutdown so the orchestrator restarts the process and the reopen
+	// re-runs recovery (or re-poisons if the fault persists = deliberate fail-stop).
+	OnPoison func(err error)
+
 	// Now and Usage are injection seams for tests; nil uses time.Now and storage.Statfs.
 	Now   func() time.Time
 	Usage func(path string) (storage.DiskUsage, error)
@@ -121,6 +131,16 @@ type Metrics interface {
 	SetDiskPressure(level Level)
 	// IncShedWrites counts one write shed because of disk pressure.
 	IncShedWrites()
+	// SetPoisoned records whether the storage engine has fail-stopped (0/1 gauge).
+	SetPoisoned(poisoned bool)
+}
+
+// Poisoner reports whether the storage engine has fail-stopped after a durability failure. The
+// wavesdb-backed store (storage.WavesdbStore) satisfies it; the in-memory store does not (it cannot
+// lose durability), so the monitor simply skips the probe when no Poisoner is configured.
+type Poisoner interface {
+	// Poisoned returns true once the engine has rejected writes after an fsync/flush/manifest failure.
+	Poisoned() (bool, error)
 }
 
 // Monitor watches a volume's free space and exposes an atomic pressure flag the write path checks. It
@@ -130,7 +150,10 @@ type Monitor struct {
 	cfg     Config
 	metrics Metrics
 
-	level atomic.Int32 // current Level
+	level    atomic.Int32 // current Level
+	poisoned atomic.Bool  // engine fail-stopped (terminal; never cleared)
+
+	poisonOnce sync.Once // guards the single OnPoison callback
 
 	mu       sync.Mutex
 	lastFree float64 // last observed free fraction, for the operator view
@@ -183,9 +206,37 @@ func (m *Monitor) Stop() {
 	<-m.doneCh
 }
 
+// checkPoison probes the engine fail-stop signal. Poison is terminal, so once observed it latches: the
+// flag stays set, the metric stays at 1, and OnPoison fires exactly once. A probe error is treated as
+// not-poisoned (we cannot assert a fail-stop we could not confirm) and simply recorded via lastErr.
+func (m *Monitor) checkPoison() {
+	if m.cfg.Poisoner == nil || m.poisoned.Load() {
+		return
+	}
+	poisoned, err := m.cfg.Poisoner.Poisoned()
+	if err != nil {
+		m.mu.Lock()
+		m.lastErr = err
+		m.mu.Unlock()
+		return
+	}
+	if !poisoned {
+		return
+	}
+	m.poisoned.Store(true)
+	if m.metrics != nil {
+		m.metrics.SetPoisoned(true)
+	}
+	if m.cfg.OnPoison != nil {
+		m.poisonOnce.Do(func() { m.cfg.OnPoison(storage.ErrPoisoned) })
+	}
+}
+
 // Sample reads the volume once and updates the level with hysteresis. Exposed so tests can drive the
-// state machine deterministically without the ticker. Returns the level after this sample.
+// state machine deterministically without the ticker. Returns the level after this sample. It also
+// probes the engine fail-stop signal (poison), which latches independently of the disk level.
 func (m *Monitor) Sample() Level {
+	m.checkPoison()
 	u, err := m.cfg.Usage(m.cfg.Path)
 	if err != nil {
 		// A Statfs error must not engage the shed (we cannot reason about free space, and gating writes on
@@ -264,20 +315,26 @@ func (m *Monitor) currentLevel() Level { return Level(m.level.Load()) }
 // Level returns the current disk-pressure level (atomic read, no lock).
 func (m *Monitor) Level() Level { return m.currentLevel() }
 
-// UnderPressure reports whether application writes should be shed right now (pressure OR critical). This
-// is the single check the write path calls; it is an atomic load, safe on the hot path.
-func (m *Monitor) UnderPressure() bool { return m.currentLevel() != LevelNone }
+// UnderPressure reports whether application writes should be shed right now: disk pressure OR critical
+// OR the engine has fail-stopped (poison). This is the single check the write path calls; it is an
+// atomic load, safe on the hot path. A poisoned engine cannot durably write, so shedding is correct
+// even before the node completes its graceful exit.
+func (m *Monitor) UnderPressure() bool { return m.currentLevel() != LevelNone || m.poisoned.Load() }
+
+// Poisoned reports whether the engine has fail-stopped (terminal until process restart).
+func (m *Monitor) Poisoned() bool { return m.poisoned.Load() }
 
 // Status is an operator-facing snapshot of the monitor.
 type Status struct {
 	Level        Level
 	FreeFraction float64
+	Poisoned     bool
 	LastErr      error
 }
 
 // Status returns the current monitor snapshot for the operator view.
 func (m *Monitor) Status() Status {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	return Status{Level: m.currentLevel(), FreeFraction: m.lastFree, LastErr: m.lastErr}
+	return Status{Level: m.currentLevel(), FreeFraction: m.lastFree, Poisoned: m.poisoned.Load(), LastErr: m.lastErr}
 }

internal/health/diskmonitor_test.go

@@ -148,10 +148,12 @@ func TestMonitorStatfsErrorDoesNotShed(t *testing.T) {
 
 // countingMetrics is a thread-safe Metrics sink for the test.
 type countingMetrics struct {
-	mu    sync.Mutex
-	level Level
-	setN  int
-	shedN int64
+	mu       sync.Mutex
+	level    Level
+	setN     int
+	shedN    int64
+	poisonN  int64
+	poisoned bool
 }
 
 func (c *countingMetrics) SetDiskPressure(l Level) {
@@ -160,6 +162,12 @@ func (c *countingMetrics) SetDiskPressure(l Level) {
 	c.mu.Unlock()
 }
 func (c *countingMetrics) IncShedWrites() { atomic.AddInt64(&c.shedN, 1) }
+func (c *countingMetrics) SetPoisoned(p bool) {
+	c.mu.Lock()
+	c.poisoned = p
+	c.mu.Unlock()
+	atomic.AddInt64(&c.poisonN, 1)
+}
 
 func TestMonitorMetricsOnTransition(t *testing.T) {
 	f := &fakeUsage{total: 1000}

internal/health/metrics.go

@@ -3,16 +3,18 @@ package health
 import "github.com/prometheus/client_golang/prometheus"
 
 // PromMetrics is the Prometheus-backed Metrics sink for the disk-pressure monitor. It exposes a
-// disk_pressure gauge (0=none, 1=pressure, 2=critical) and a shed-writes counter.
+// disk_pressure gauge (0=none, 1=pressure, 2=critical), a shed-writes counter, and a storage_poisoned
+// gauge (0/1) set when the engine fail-stops after a durability failure.
 type PromMetrics struct {
 	pressure prometheus.Gauge
 	shed     prometheus.Counter
+	poisoned prometheus.Gauge
 }
 
 var _ Metrics = (*PromMetrics)(nil)
 
-// NewPromMetrics registers the disk-pressure collectors against reg and returns the sink. The gauge
-// starts at 0 (no pressure).
+// NewPromMetrics registers the disk-pressure collectors against reg and returns the sink. The gauges
+// start at 0 (no pressure, not poisoned).
 func NewPromMetrics(reg prometheus.Registerer) *PromMetrics {
 	m := &PromMetrics{
 		pressure: prometheus.NewGauge(prometheus.GaugeOpts{
@@ -23,8 +25,12 @@ func NewPromMetrics(reg prometheus.Registerer) *PromMetrics {
 			Name: "wavespan_disk_pressure_shed_writes_total",
 			Help: "Writes shed at admission because the storage volume was under disk pressure.",
 		}),
+		poisoned: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "wavespan_storage_poisoned",
+			Help: "1 when the storage engine has fail-stopped after a durability failure (writes rejected); 0 otherwise.",
+		}),
 	}
-	reg.MustRegister(m.pressure, m.shed)
+	reg.MustRegister(m.pressure, m.shed, m.poisoned)
 	return m
 }
 
@@ -33,3 +39,12 @@ func (m *PromMetrics) SetDiskPressure(level Level) { m.pressure.Set(float64(leve
 
 // IncShedWrites counts one shed write.
 func (m *PromMetrics) IncShedWrites() { m.shed.Inc() }
+
+// SetPoisoned records whether the storage engine has fail-stopped.
+func (m *PromMetrics) SetPoisoned(poisoned bool) {
+	if poisoned {
+		m.poisoned.Set(1)
+		return
+	}
+	m.poisoned.Set(0)
+}