Pebble based Storage Engine

.github/workflows/ci.yaml

@@ -14,7 +14,12 @@ jobs:
         uses: golangci/golangci-lint-action@v9
         with:
           skip-cache: true
-          version: latest
+          # Pinned so lint findings don't drift with new golangci-lint
+          # releases. New gosec rules (G702/G118/G122/G703/G704) landed
+          # in v2.10+, breaking PRs that were clean against earlier
+          # versions. Bump deliberately, fixing any new findings in the
+          # same PR.
+          version: v2.12.2
           args: --timeout=3m
   go-test:
     strategy:

.github/workflows/main.yaml

@@ -33,7 +33,10 @@ jobs:
         uses: golangci/golangci-lint-action@v9
         with:
           skip-cache: true
-          version: latest
+          # Pinned. See ci.yaml note: a `latest` golangci-lint version
+          # bump can retroactively break PRs and main when new rules
+          # land. Update deliberately and fix new findings in the same PR.
+          version: v2.12.2
           args: --timeout=3m
   go-test:
     strategy:

buf.gen.yaml

@@ -13,3 +13,8 @@ plugins:
   - remote: buf.build/bufbuild/validate-go:v1.2.1
     out: pb
     opt: paths=source_relative
+  - local: protoc-gen-go-vtproto
+    out: pb
+    opt:
+      - paths=source_relative
+      - features=marshal+unmarshal+size

cmd/baton-fixture-gen/main.go

@@ -0,0 +1,126 @@
+//go:build batonsdkv2
+
+// Command baton-fixture-gen generates synthetic hyper-scale c1z3
+// fixtures for the storage-engine benchmark + equivalence harnesses
+// (RFC 0004 Stack 5).
+//
+// Usage:
+//
+//	baton-fixture-gen \
+//	    -out /tmp/fixture-1m-grants \
+//	    -grants 1000000 \
+//	    -entitlements 1000 \
+//	    -principals 50000 \
+//	    -sync-id 00000000000000000000000001
+//
+// The output is a fresh Pebble directory containing one sync's worth
+// of GrantRecord rows + indexes. Run baton-storage-bench against this
+// fixture to time the canonical reader workloads.
+//
+// Determinism: with the same -seed the same fixture is produced
+// byte-for-byte (modulo Pebble's compaction non-determinism, which
+// affects file layout but not logical content).
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"math/rand"
+	"os"
+	"strconv"
+	"time"
+
+	v3 "github.com/conductorone/baton-sdk/pb/c1/storage/v3"
+	enginepkg "github.com/conductorone/baton-sdk/pkg/dotc1z/engine/pebble"
+)
+
+func main() {
+	out := flag.String("out", "", "destination directory (must not exist)")
+	grants := flag.Int("grants", 100_000, "total number of grants to generate")
+	entitlements := flag.Int("entitlements", 100, "number of distinct entitlements")
+	principals := flag.Int("principals", 1_000, "number of distinct principals")
+	syncID := flag.String("sync-id", "", "sync ID to use (default: 25-char KSUID)")
+	seed := flag.Int64("seed", 1, "PRNG seed for deterministic generation")
+	flag.Parse()
+
+	if *out == "" {
+		log.Fatal("baton-fixture-gen: -out is required")
+	}
+	if _, err := os.Stat(*out); err == nil {
+		log.Fatalf("baton-fixture-gen: %s already exists", *out)
+	}
+
+	if *syncID == "" {
+		*syncID = "2N7tQT4Yx8z" + strconv.FormatInt(time.Now().Unix(), 10)
+		// Pad/truncate to 27 chars (KSUID string length).
+		if len(*syncID) > 27 {
+			*syncID = (*syncID)[:27]
+		}
+		for len(*syncID) < 27 {
+			*syncID += "0"
+		}
+	}
+
+	if err := run(*out, *syncID, *grants, *entitlements, *principals, *seed); err != nil {
+		log.Fatalf("baton-fixture-gen: %v", err)
+	}
+}
+
+func run(outDir, syncID string, totalGrants, numEnt, numPrinc int, seed int64) error {
+	ctx := context.Background()
+	e, err := enginepkg.Open(ctx, outDir)
+	if err != nil {
+		return fmt.Errorf("open engine: %w", err)
+	}
+	defer e.Close()
+
+	if err := e.SetCurrentSync(syncID); err != nil {
+		return fmt.Errorf("SetCurrentSync: %w", err)
+	}
+
+	rng := rand.New(rand.NewSource(seed)) //nolint:gosec // deterministic fixture generation; not security-sensitive
+	progressEvery := totalGrants / 20
+	if progressEvery == 0 {
+		progressEvery = 1
+	}
+
+	start := time.Now()
+	for i := 0; i < totalGrants; i++ {
+		entID := fmt.Sprintf("ent-%06d", rng.Intn(numEnt))
+		principalID := fmt.Sprintf("user-%08d", rng.Intn(numPrinc))
+		ext := fmt.Sprintf("grant-%010d", i)
+
+		r := v3.GrantRecord_builder{
+			SyncId:     syncID,
+			ExternalId: ext,
+			Entitlement: v3.EntitlementRef_builder{
+				ResourceTypeId: "app",
+				ResourceId:     fmt.Sprintf("app-%d", rng.Intn(10)),
+				EntitlementId:  entID,
+			}.Build(),
+			Principal: v3.PrincipalRef_builder{
+				ResourceTypeId: "user",
+				ResourceId:     principalID,
+			}.Build(),
+		}.Build()
+
+		if err := e.PutGrantRecord(ctx, r); err != nil {
+			return fmt.Errorf("PutGrantRecord[%d]: %w", i, err)
+		}
+
+		if i%progressEvery == 0 {
+			elapsed := time.Since(start)
+			rps := float64(i) / elapsed.Seconds()
+			fmt.Fprintf(os.Stderr, "[%5.1f%%] %d/%d grants  %.0f r/s  elapsed=%v\n",
+				100*float64(i)/float64(totalGrants), i, totalGrants, rps, elapsed.Round(time.Millisecond))
+		}
+	}
+
+	elapsed := time.Since(start)
+	rps := float64(totalGrants) / elapsed.Seconds()
+	fmt.Fprintf(os.Stderr, "done: %d grants in %v (%.0f r/s)\n",
+		totalGrants, elapsed.Round(time.Millisecond), rps)
+	return nil
+}

cmd/baton-storage-bench/main.go

@@ -0,0 +1,187 @@
+//go:build batonsdkv2
+
+// Command baton-storage-bench is the canonical benchmark harness for
+// the v3 Pebble storage engine (RFC 0004 §6, Stack 5).
+//
+// Targets the goal table from RFC §6:
+//
+//	G1  Bulk write grants from a single sync_run
+//	G2  Random Get by primary key
+//	G3  List by entitlement
+//	G4  List by principal
+//	G5  Full-sync scan
+//	G6  Cold open + Checkpoint
+//	G7  Cross-engine compaction (delegated to synccompactor/pebble)
+//	G8  Memory ceiling under burst
+//	G9  Save → Open → Read roundtrip
+//	G10 Engine open count for the multi-engine ReaderCache
+//
+// Stack 5 MVP implements G1–G5 (the read/write hot path). G6–G10 land
+// alongside Stack 5 follow-up commits as the canonical fixture and
+// cmd/c1z-bench expand.
+//
+// Usage:
+//
+//	# generate a fixture first:
+//	baton-fixture-gen -out /tmp/fix-1m -grants 1000000 \
+//	    -entitlements 1000 -principals 50000
+//
+//	# then bench:
+//	baton-storage-bench -input /tmp/fix-1m -sync-id ... -bench G3 -duration 30s
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"math/rand"
+	"os"
+	"time"
+
+	v3 "github.com/conductorone/baton-sdk/pb/c1/storage/v3"
+	enginepkg "github.com/conductorone/baton-sdk/pkg/dotc1z/engine/pebble"
+)
+
+func main() {
+	input := flag.String("input", "", "input Pebble directory (built by baton-fixture-gen)")
+	syncID := flag.String("sync-id", "", "sync_id to bench against")
+	benchName := flag.String("bench", "G3", "benchmark to run: G2|G3|G4|G5")
+	duration := flag.Duration("duration", 10*time.Second, "minimum total benchmark duration")
+	entitlements := flag.Int("entitlements", 100, "matches the fixture's -entitlements")
+	principals := flag.Int("principals", 1000, "matches the fixture's -principals")
+	flag.Parse()
+
+	if *input == "" {
+		log.Fatal("baton-storage-bench: -input is required")
+	}
+	if *syncID == "" {
+		log.Fatal("baton-storage-bench: -sync-id is required")
+	}
+
+	ctx := context.Background()
+	e, err := enginepkg.Open(ctx, *input, enginepkg.WithReadOnly(true))
+	if err != nil {
+		log.Fatalf("open engine: %v", err)
+	}
+	defer e.Close()
+	if err := e.SetCurrentSync(*syncID); err != nil {
+		_ = e.Close()
+		fmt.Fprintf(os.Stderr, "SetCurrentSync: %v\n", err)
+		os.Exit(1) //nolint:gocritic // close already invoked above
+	}
+
+	switch *benchName {
+	case "G2":
+		runG2(ctx, e, *syncID, *duration)
+	case "G3":
+		runG3(ctx, e, *syncID, *entitlements, *duration)
+	case "G4":
+		runG4(ctx, e, *syncID, *principals, *duration)
+	case "G5":
+		runG5(ctx, e, *syncID, *duration)
+	default:
+		log.Fatalf("unknown bench %q (G2|G3|G4|G5 supported)", *benchName)
+	}
+}
+
+// G2 — random Get by primary key.
+func runG2(ctx context.Context, e *enginepkg.Engine, syncID string, dur time.Duration) {
+	rng := rand.New(rand.NewSource(1)) //nolint:gosec // deterministic bench seed
+	deadline := time.Now().Add(dur)
+	var ops, hits, misses int64
+
+	start := time.Now()
+	for time.Now().Before(deadline) {
+		ext := fmt.Sprintf("grant-%010d", rng.Intn(1_000_000))
+		_, err := e.GetGrantRecord(ctx, syncID, ext)
+		ops++
+		if err == nil {
+			hits++
+		} else {
+			misses++
+		}
+	}
+	elapsed := time.Since(start)
+	report("G2 Get-by-PK", elapsed, ops, hits, misses)
+}
+
+// G3 — list by entitlement.
+func runG3(ctx context.Context, e *enginepkg.Engine, syncID string, numEnt int, dur time.Duration) {
+	rng := rand.New(rand.NewSource(2)) //nolint:gosec // deterministic bench seed
+	deadline := time.Now().Add(dur)
+	var ops, rows int64
+
+	start := time.Now()
+	for time.Now().Before(deadline) {
+		entID := fmt.Sprintf("ent-%06d", rng.Intn(numEnt))
+		var got int
+		err := e.IterateGrantsByEntitlement(ctx, syncID, entID, func(*v3.GrantRecord) bool {
+			got++
+			return true
+		})
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "iter error: %v\n", err)
+			os.Exit(1)
+		}
+		ops++
+		rows += int64(got)
+	}
+	elapsed := time.Since(start)
+	report("G3 ListByEntitlement", elapsed, ops, rows, 0)
+}
+
+// G4 — list by principal.
+func runG4(ctx context.Context, e *enginepkg.Engine, syncID string, numPrinc int, dur time.Duration) {
+	rng := rand.New(rand.NewSource(3)) //nolint:gosec // deterministic bench seed
+	deadline := time.Now().Add(dur)
+	var ops, rows int64
+
+	start := time.Now()
+	for time.Now().Before(deadline) {
+		pid := fmt.Sprintf("user-%08d", rng.Intn(numPrinc))
+		var got int
+		err := e.IterateGrantsByPrincipal(ctx, syncID, "user", pid, func(*v3.GrantRecord) bool {
+			got++
+			return true
+		})
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "iter error: %v\n", err)
+			os.Exit(1)
+		}
+		ops++
+		rows += int64(got)
+	}
+	elapsed := time.Since(start)
+	report("G4 ListByPrincipal", elapsed, ops, rows, 0)
+}
+
+// G5 — full-sync scan.
+func runG5(ctx context.Context, e *enginepkg.Engine, syncID string, dur time.Duration) {
+	deadline := time.Now().Add(dur)
+	var ops, rows int64
+
+	start := time.Now()
+	for time.Now().Before(deadline) {
+		var got int
+		err := e.IterateGrantsBySync(ctx, syncID, func(*v3.GrantRecord) bool {
+			got++
+			return true
+		})
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "iter error: %v\n", err)
+			os.Exit(1)
+		}
+		ops++
+		rows += int64(got)
+	}
+	elapsed := time.Since(start)
+	report("G5 FullSyncScan", elapsed, ops, rows, 0)
+}
+
+func report(name string, elapsed time.Duration, ops, a, b int64) {
+	opsPerSec := float64(ops) / elapsed.Seconds()
+	nsPerOp := float64(elapsed.Nanoseconds()) / float64(ops)
+	fmt.Fprintf(os.Stdout, "%s: %d ops in %v  %.0f ops/s  %.0f ns/op  a=%d b=%d\n",
+		name, ops, elapsed.Round(time.Millisecond), opsPerSec, nsPerOp, a, b)
+}