cabbage/rvbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: bootstrap durable server store

Browse Source
This commit is contained in:
cabbage
2026-08-31 09:11:58 +00:00
parent 392129c253
commit 8a88a72e62
13 changed files with 822 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/coverage.toml
+36
View File
@@ -165,3 +165,39 @@ id = "HP-WINCTX-02"
layer = "integration"
status = "blocked_native_windows"
tests = []
[[requirements]]
id = "HP-STORE-01"
layer = "integration"
status = "implemented"
tests = ["test/integration/store/store_integration_test.go:TestRealSQLiteInitializationAndRestart_HP_STORE_01"]
[[requirements]]
id = "BH-STORE-01"
layer = "integration"
status = "implemented"
tests = ["test/integration/store/store_integration_test.go:TestSchemaRejectsInvalidIDsAndForeignKeys_BH_STORE_01"]
[[requirements]]
id = "BH-STORE-02"
layer = "integration"
status = "implemented"
tests = ["test/integration/store/store_integration_test.go:TestUnsafePathsAndCancelledOpen_BH_STORE_02"]
[[requirements]]
id = "BH-STORE-03"
layer = "unit"
status = "implemented"
tests = ["test/harness/harness_test.go:TestBoundedSuiteLogAndFailedRunRecovery_BH_STORE_03"]
[[requirements]]
id = "RACE-STORE-01"
layer = "integration"
status = "implemented"
tests = ["test/integration/store/store_integration_test.go:TestSingleInstanceLockAndConcurrentClose_RACE_STORE_01"]
[[requirements]]
id = "REC-STORE-01"
layer = "integration"
status = "implemented"
tests = ["test/integration/store/store_integration_test.go:TestMigrationChecksumMismatchPreventsOpen_REC_STORE_01"]
test/harness/harness.go
+65 -7
View File
@@ -16,6 +16,7 @@ import (
"os/exec"
"path/filepath"
"regexp"
"strconv"
"strings"
"time"
@@ -26,6 +27,7 @@ import (
const (
manifestVersion = 1
repositoryID = "rvbox"
maxSuiteLogSize = 1 << 20
)
var runIDPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,63}$`)
@@ -124,8 +126,8 @@ func (h *harness) integration(ctx context.Context, args []string) error {
if err := flags.Parse(args); err != nil {
return err
}
if *suite != "sample" {
return fmt.Errorf("suite %q is not implemented yet; available: sample", *suite)
if *suite != "sample" && *suite != "store" {
return fmt.Errorf("suite %q is not implemented yet; available: sample, store", *suite)
}
var current *manifest
@@ -141,7 +143,7 @@ func (h *harness) integration(ctx context.Context, args []string) error {
if current.Layer != "integration" || current.Suite != *suite {
return errors.New("run layer/suite does not match resume request")
}
if current.Phase != "ready" && current.Phase != "interrupted" && current.Phase != "stopped" && current.Phase != "running" {
if current.Phase != "ready" && current.Phase != "interrupted" && current.Phase != "stopped" && current.Phase != "running" && current.Phase != "failed" {
return fmt.Errorf("run in phase %q is not resumable; recover or reuse it first", current.Phase)
}
} else {
@@ -151,7 +153,7 @@ func (h *harness) integration(ctx context.Context, args []string) error {
}
}
fmt.Fprintln(h.out, current.RunID)
if err := h.transition(current, "running", "sample-start", "sample integration run started"); err != nil {
if err := h.transition(current, "running", *suite+"-start", *suite+" integration run started"); err != nil {
return err
}
select {
@@ -160,10 +162,38 @@ func (h *harness) integration(ctx context.Context, args []string) error {
return ctx.Err()
default:
}
if err := h.appendJournal(current.RunID, journalEntry{At: h.now(), Step: "invariant", Status: "passed", Detail: "manifest ownership and journal durability verified"}); err != nil {
if *suite == "sample" {
if err := h.appendJournal(current.RunID, journalEntry{At: h.now(), Step: "invariant", Status: "passed", Detail: "manifest ownership and journal durability verified"}); err != nil {
return err
}
} else if err := h.runStoreSuite(ctx, current); err != nil {
_ = h.transition(current, "failed", "store-failed", err.Error())
return err
}
return h.transition(current, "completed", "sample-complete", "sample integration run completed")
return h.transition(current, "completed", *suite+"-complete", *suite+" integration run completed")
}
func (h *harness) runStoreSuite(ctx context.Context, current *manifest) error {
if err := h.appendJournal(current.RunID, journalEntry{At: h.now(), Step: "store-real-sqlite", Status: "running", Detail: "running real SQLite/WAL and filesystem cases"}); err != nil {
return err
}
artifactDir := filepath.Join(h.runDir(current.RunID), "artifacts")
if err := os.MkdirAll(artifactDir, 0o700); err != nil {
return err
}
capture := &limitedCapture{limit: maxSuiteLogSize}
command := exec.CommandContext(ctx, "go", "test", "-count=1", "-tags=integration", "-shuffle="+strconv.FormatInt(current.Seed, 10), "-timeout=2m", "./test/integration/store")
command.Stdout = capture
command.Stderr = capture
err := command.Run()
logPath := filepath.Join(artifactDir, "suite.log")
if writeErr := atomicWrite(logPath, capture.Bytes(), 0o600); writeErr != nil {
return writeErr
}
if err != nil {
return fmt.Errorf("store suite failed (bounded log %s): %w", logPath, err)
}
return h.appendJournal(current.RunID, journalEntry{At: h.now(), Step: "store-real-sqlite", Status: "passed", Detail: "real SQLite/WAL and filesystem cases passed"})
}
func (h *harness) environmentCommand(command string, args []string) error {
@@ -195,7 +225,7 @@ func (h *harness) environmentCommand(command string, args []string) error {
case "collect":
return h.collect(current)
case "recover":
if current.Phase != "interrupted" && current.Phase != "stopped" && current.Phase != "running" {
if current.Phase != "interrupted" && current.Phase != "stopped" && current.Phase != "running" && current.Phase != "failed" {
return fmt.Errorf("run in phase %q does not need recovery", current.Phase)
}
return h.transition(current, "ready", "recover", "run recovered and ready to resume")
@@ -215,6 +245,34 @@ func (h *harness) environmentCommand(command string, args []string) error {
}
}
type limitedCapture struct {
data []byte
limit int
truncated bool
}
func (capture *limitedCapture) Write(data []byte) (int, error) {
written := len(data)
remaining := capture.limit - len(capture.data)
if remaining > 0 {
if len(data) > remaining {
data = data[:remaining]
}
capture.data = append(capture.data, data...)
}
if written > remaining {
capture.truncated = true
}
return written, nil
}
func (capture *limitedCapture) Bytes() []byte {
if !capture.truncated {
return capture.data
}
return append(append([]byte(nil), capture.data...), []byte("\n[output truncated by RVBox test harness]\n")...)
}
func (h *harness) create(requestedID, layer, suite string) (*manifest, error) {
if requestedID == "" {
id, err := domain.NewUUIDv7()
test/harness/harness_test.go
+28
View File
@@ -112,6 +112,34 @@ func TestIntegrationResumeValidation_HP_CFG_01(t *testing.T) {
}
}
func TestBoundedSuiteLogAndFailedRunRecovery_BH_STORE_03(t *testing.T) {
t.Parallel()
capture := &limitedCapture{limit: 4}
if written, err := capture.Write([]byte("123456")); err != nil || written != 6 {
t.Fatalf("Write = (%d, %v)", written, err)
}
if got := string(capture.Bytes()); !strings.HasPrefix(got, "1234\n") || !strings.Contains(got, "truncated") {
t.Fatalf("bounded output = %q", got)
}
h := &harness{root: t.TempDir(), now: time.Now, out: &bytes.Buffer{}}
current, err := h.create("failed-store", "integration", "store")
if err != nil {
t.Fatal(err)
}
if err := h.transition(current, "failed", "store-failed", "injected"); err != nil {
t.Fatal(err)
}
if err := h.environmentCommand("recover", []string{"--run-id", current.RunID}); err != nil {
t.Fatalf("recover failed run: %v", err)
}
loaded, err := h.load(current.RunID)
if err != nil || loaded.Phase != "ready" {
t.Fatalf("recovered run = (%+v, %v)", loaded, err)
}
}
func TestCoverageInventoryReferencesExistingTests_HP_CFG_01(t *testing.T) {
t.Parallel()
test/integration/store/store_integration_test.go
+236
View File
@@ -0,0 +1,236 @@
//go:build integration
package store_test
import (
"context"
"database/sql"
"errors"
"os"
"path/filepath"
"runtime"
"sort"
"strings"
"sync"
"testing"
"time"
"github.com/rvbox/rvbox/internal/server/store"
)
const busyTimeout = 2 * time.Second
func TestRealSQLiteInitializationAndRestart_HP_STORE_01(t *testing.T) {
t.Parallel()
dataDir := filepath.Join(t.TempDir(), "state")
opened := openStore(t, dataDir)
for _, directory := range []string{dataDir, filepath.Join(dataDir, "segments"), filepath.Join(dataDir, "audit")} {
info, err := os.Stat(directory)
if err != nil {
t.Fatal(err)
}
if runtime.GOOS != "windows" && info.Mode().Perm() != 0o700 {
t.Fatalf("%s mode = %04o, want 0700", directory, info.Mode().Perm())
}
}
for _, file := range []string{"rvbox.db", "server.lock"} {
info, err := os.Stat(filepath.Join(dataDir, file))
if err != nil {
t.Fatal(err)
}
if runtime.GOOS != "windows" && info.Mode().Perm() != 0o600 {
t.Fatalf("%s mode = %04o, want 0600", file, info.Mode().Perm())
}
}
assertPragma(t, opened.DB(), "journal_mode", "wal")
assertPragma(t, opened.DB(), "foreign_keys", "1")
assertPragma(t, opened.DB(), "synchronous", "2")
assertPragma(t, opened.DB(), "busy_timeout", "2000")
rows, err := opened.DB().Query(`SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name`)
if err != nil {
t.Fatal(err)
}
var names []string
for rows.Next() {
var name string
if err := rows.Scan(&name); err != nil {
t.Fatal(err)
}
names = append(names, name)
}
if err := rows.Close(); err != nil {
t.Fatal(err)
}
want := []string{"audit_events", "clients", "command_events", "command_payloads", "command_tombstones", "commands", "control_mutations", "output_segments", "output_truncations", "schema_migrations", "sessions", "stdin_writes", "storage_incidents", "takeover_authorizations"}
sort.Strings(want)
if strings.Join(names, ",") != strings.Join(want, ",") {
t.Fatalf("tables = %v, want %v", names, want)
}
var migrationCount int
if err := opened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 1 {
t.Fatalf("migration count = %d, err = %v", migrationCount, err)
}
if err := opened.Close(); err != nil {
t.Fatal(err)
}
reopened := openStore(t, dataDir)
if err := reopened.DB().QueryRow(`SELECT count(*) FROM schema_migrations`).Scan(&migrationCount); err != nil || migrationCount != 1 {
t.Fatalf("reopened migration count = %d, err = %v", migrationCount, err)
}
if err := reopened.Close(); err != nil {
t.Fatal(err)
}
}
func TestSingleInstanceLockAndConcurrentClose_RACE_STORE_01(t *testing.T) {
t.Parallel()
dataDir := filepath.Join(t.TempDir(), "state")
first := openStore(t, dataDir)
if _, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrAlreadyOpen) {
t.Fatalf("second Open error = %v, want ErrAlreadyOpen", err)
}
var wait sync.WaitGroup
for range 8 {
wait.Add(1)
go func() {
defer wait.Done()
if err := first.Close(); err != nil && !errors.Is(err, sql.ErrConnDone) {
t.Errorf("Close: %v", err)
}
}()
}
wait.Wait()
reopened := openStore(t, dataDir)
if err := reopened.Close(); err != nil {
t.Fatal(err)
}
}
func TestSchemaRejectsInvalidIDsAndForeignKeys_BH_STORE_01(t *testing.T) {
t.Parallel()
opened := openStore(t, filepath.Join(t.TempDir(), "state"))
defer opened.Close()
_, err := opened.DB().Exec(`INSERT INTO clients (
client_id, platform, architecture, daemon_version, daemon_cwd, supported_shells, capabilities, client_instance_id
) VALUES ('client-a', 3, 'amd64', 'test', 'C:\\work', x'', x'', x'01')`)
if err == nil {
t.Fatal("invalid client instance UUID was accepted")
}
_, err = opened.DB().Exec(`INSERT INTO sessions (
session_id, client_id, client_instance_id, generation, opened_at
) VALUES (?, 'missing-client', ?, 1, 1)`, bytesOf(16, 1), bytesOf(16, 2))
if err == nil || !strings.Contains(strings.ToLower(err.Error()), "foreign key") {
t.Fatalf("foreign-key insert error = %v", err)
}
}
func TestUnsafePathsAndCancelledOpen_BH_STORE_02(t *testing.T) {
t.Parallel()
if _, err := store.Open(context.Background(), store.Options{DataDir: "relative", BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
t.Fatalf("relative path error = %v", err)
}
if _, err := store.Open(context.Background(), store.Options{DataDir: string(filepath.Separator), BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
t.Fatalf("root path error = %v", err)
}
permissive := filepath.Join(t.TempDir(), "permissive")
if err := os.Mkdir(permissive, 0o755); err != nil {
t.Fatal(err)
}
if _, err := store.Open(context.Background(), store.Options{DataDir: permissive, BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
t.Fatalf("permissive path error = %v", err)
}
if runtime.GOOS != "windows" {
target := filepath.Join(t.TempDir(), "target")
if err := os.Mkdir(target, 0o700); err != nil {
t.Fatal(err)
}
link := filepath.Join(t.TempDir(), "linked")
if err := os.Symlink(target, link); err != nil {
t.Fatal(err)
}
if _, err := store.Open(context.Background(), store.Options{DataDir: link, BusyTimeout: busyTimeout}); !errors.Is(err, store.ErrUnsafeDataDirectory) {
t.Fatalf("symlink path error = %v", err)
}
}
cancelled, cancel := context.WithCancel(context.Background())
cancel()
dataDir := filepath.Join(t.TempDir(), "cancelled")
if _, err := store.Open(cancelled, store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}); !errors.Is(err, context.Canceled) {
t.Fatalf("cancelled Open error = %v", err)
}
opened := openStore(t, dataDir)
if err := opened.Close(); err != nil {
t.Fatal(err)
}
}
func TestMigrationChecksumMismatchPreventsOpen_REC_STORE_01(t *testing.T) {
t.Parallel()
dataDir := filepath.Join(t.TempDir(), "state")
opened := openStore(t, dataDir)
if err := opened.Close(); err != nil {
t.Fatal(err)
}
raw, err := sql.Open("sqlite", filepath.Join(dataDir, "rvbox.db"))
if err != nil {
t.Fatal(err)
}
if _, err := raw.Exec(`UPDATE schema_migrations SET checksum = 'tampered' WHERE version = 1`); err != nil {
t.Fatal(err)
}
if err := raw.Close(); err != nil {
t.Fatal(err)
}
if _, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout}); err == nil || !strings.Contains(err.Error(), "checksum mismatch") {
t.Fatalf("Open error = %v, want checksum mismatch", err)
}
// A failed open must release the process lock for offline inspection/repair.
raw, err = sql.Open("sqlite", filepath.Join(dataDir, "rvbox.db"))
if err != nil {
t.Fatal(err)
}
if err := raw.Ping(); err != nil {
t.Fatal(err)
}
if err := raw.Close(); err != nil {
t.Fatal(err)
}
}
func openStore(t *testing.T, dataDir string) *store.Store {
t.Helper()
opened, err := store.Open(context.Background(), store.Options{DataDir: dataDir, BusyTimeout: busyTimeout})
if err != nil {
t.Fatalf("Open(%s): %v", dataDir, err)
}
return opened
}
func assertPragma(t *testing.T, database *sql.DB, name, want string) {
t.Helper()
var got string
if err := database.QueryRow(`PRAGMA ` + name).Scan(&got); err != nil {
t.Fatalf("PRAGMA %s: %v", name, err)
}
if strings.ToLower(got) != want {
t.Fatalf("PRAGMA %s = %q, want %q", name, got, want)
}
}
func bytesOf(length int, value byte) []byte {
result := make([]byte, length)
for index := range result {
result[index] = value
}
return result
}