diff --git a/README.md b/README.md index cb33e24..7ee432c 100644 --- a/README.md +++ b/README.md @@ -157,6 +157,33 @@ NixOS and native development services expose `softVouchLists`, false by default. CAUTION: List owners can change members and therefore change subscribers' trust graphs. List subscriptions add stream traffic, database records, and PDS requests. Replay gaps still require explicit participant sync. +### Index capacity + +The soft record and replay-version tables use `WITHOUT ROWID` to avoid duplicate primary-key storage. +Schema version 7 rebuilds these tables atomically. It preserves records, replay tombstones, list provenance, and cached path dependencies. +The migration needs temporary disk space for the original tables, rebuilt tables, indexes, and WAL. +SQLite reuses freed pages, but the database file does not automatically shrink after the migration. +After a backup and a service stop, an operator can use SQLite `VACUUM` to reclaim unused pages. +`VACUUM` also needs temporary disk space. No automatic compaction or tombstone cleanup exists. + +The scale benchmark uses 100,000 follows, 1,000 authors, realistic DID lengths, live replay versions, and the production indexes. +On the same fixture, database pages decrease from 644 to 497 bytes per edge (23%). +Graph-load and selection allocations decrease from 162 MB to 118 MB (27%). +These figures exclude WAL files, deleted-record tombstones, cached results, and list data. Allocation totals are not peak RAM measurements. +They are fixture measurements, not capacity guarantees for the Bluesky network. + +Run the reproducible benchmarks: + +```sh +go test ./internal/storage ./internal/trust -run '^$' -bench 'Benchmark.*Scale' -benchtime=1x -benchmem +``` + +CAUTION: All-author ingestion does not make global trust queries viable at Bluesky scale. +Uncached queries still load the global graph and reject inputs that exceed the default one-million-edge limit. +Concurrent uncached queries can each allocate their own graph and traversal state. +Trust cache rows, replay tombstones, and referenced list data still grow without automatic cleanup. +The all-author mode also lacks a historical network-wide bootstrap. It indexes observed stream events, not all existing follows and blocks. + ## API ### Trust algorithms diff --git a/docs/security-review.md b/docs/security-review.md index 1df00cd..1a20416 100644 --- a/docs/security-review.md +++ b/docs/security-review.md @@ -89,6 +89,9 @@ Participant sync recovers soft records after long outages or disabled periods. Blocked-list owners control membership and can change subscribers' calculated trust. A subscription delegates this part of the graph to its owner. Blocked-list imports have bounded record counts, but total indexed list data and retained replay tombstones have no automatic cleanup. An unavailable list owner's PDS can delay a new subscription event and its stream cursor until recovery. +Schema version 7 rebuilds soft records and replay versions with compact primary-key storage. The migration needs temporary disk space. +Global graph loads remain a RAM and query-capacity limit. Concurrent uncached queries multiply these allocations. +No tombstone retention limit exists. A cleanup policy must preserve replay protection before it removes old versions. Run only one appview ingestor per database. Repository operations serialize within one UI process. Multiple UI replicas need shared refresh coordination. Live PDS OAuth interoperability still needs an account-based smoke test. diff --git a/internal/appview/service.go b/internal/appview/service.go index e3b2072..ba2ab5b 100644 --- a/internal/appview/service.go +++ b/internal/appview/service.go @@ -142,7 +142,7 @@ func (s *Service) Reputation(ctx context.Context, source, target, tag string, mo } // Attach reference text after calculation so that algorithms never select paths by comment content. -func pathComments(result trust.Result, records map[[2]string]domain.Vouch) trust.Result { +func pathComments(result trust.Result, records map[[2]string]*domain.Vouch) trust.Result { if len(result.PathScores) == 0 { return result } @@ -151,11 +151,16 @@ func pathComments(result trust.Result, records map[[2]string]domain.Vouch) trust result.PathScores[0].Comment = "" result.PathScores[0].Source = "" result.PathScores[0].ListURI = "" + var empty domain.Vouch for i := 1; i < len(result.PathScores); i++ { pair := [2]string{result.PathScores[i-1].DID, result.PathScores[i].DID} - result.PathScores[i].Comment = records[pair].Comment - result.PathScores[i].Source = records[pair].Source - result.PathScores[i].ListURI = records[pair].ListURI + record := records[pair] + if record == nil { + record = &empty + } + result.PathScores[i].Comment = record.Comment + result.PathScores[i].Source = record.Source + result.PathScores[i].ListURI = record.ListURI } return result } diff --git a/internal/appview/trust_cache.go b/internal/appview/trust_cache.go index 886cd09..9a2de01 100644 --- a/internal/appview/trust_cache.go +++ b/internal/appview/trust_cache.go @@ -24,7 +24,7 @@ type trustReader struct { tag string mode domain.Mode edges []domain.Vouch - records map[[2]string]domain.Vouch + records map[[2]string]*domain.Vouch revision int64 loaded bool } @@ -114,15 +114,17 @@ func cacheKey(descriptor trust.Descriptor, input trust.Input, tag string, mode d return hex.EncodeToString(digest[:]), nil } -func pathRecords(edges []domain.Vouch) map[[2]string]domain.Vouch { - records := make(map[[2]string]domain.Vouch, len(edges)) - for _, edge := range edges { +func pathRecords(edges []domain.Vouch) map[[2]string]*domain.Vouch { + // The selected edges remain immutable for the lifetime of this query. + records := make(map[[2]string]*domain.Vouch, len(edges)) + for index := range edges { + edge := &edges[index] records[[2]string{edge.Author, edge.Subject}] = edge } return records } -func pathDependencies(result trust.Result, records map[[2]string]domain.Vouch) []trust.Dependency { +func pathDependencies(result trust.Result, records map[[2]string]*domain.Vouch) []trust.Dependency { dependencies := make([]trust.Dependency, 0, len(result.PathScores)) for i := 1; i < len(result.PathScores); i++ { pair := [2]string{result.PathScores[i-1].DID, result.PathScores[i].DID} diff --git a/internal/storage/compact_test.go b/internal/storage/compact_test.go new file mode 100644 index 0000000..ddb4b8a --- /dev/null +++ b/internal/storage/compact_test.go @@ -0,0 +1,87 @@ +package storage + +import ( + "context" + "database/sql" + "path/filepath" + "strings" + "testing" + "time" + + "provenance/internal/domain" + "provenance/internal/storage/db" + "provenance/internal/trust" +) + +func TestCompactMigration(t *testing.T) { + ctx := context.Background() + path := filepath.Join(t.TempDir(), "version6.db") + conn, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + for _, statement := range []string{"PRAGMA foreign_keys=ON", schema, strings.ReplaceAll(softSchema, ") WITHOUT ROWID;", ");"), strings.ReplaceAll(listsSchema, ") WITHOUT ROWID;", ");"), "PRAGMA user_version=6"} { + if _, err := conn.Exec(statement); err != nil { + t.Fatal(err) + } + } + legacy := &SQLite{conn: conn, queries: db.New(conn), softMode: domain.SoftParticipantLists} + if err := legacy.Join(ctx, "did:plc:a"); err != nil { + t.Fatal(err) + } + follow := domain.Vouch{Author: "did:plc:a", Subject: "did:plc:b", Source: domain.FollowCollection, RecordKey: "follow", Score: .5, CreatedAt: listDate} + if err := legacy.ReplaceWithLists(ctx, follow.Author, []domain.Vouch{follow}, listFixture(), time.UnixMicro(100)); err != nil { + t.Fatal(err) + } + if err := legacy.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: follow.Author, Source: follow.Source, RecordKey: "deleted", TimeUS: 300}); err != nil { + t.Fatal(err) + } + _, revision, err := legacy.TrustGraph(ctx, "", domain.Default, 100) + if err != nil { + t.Fatal(err) + } + score := -.5 + deps := []trust.Dependency{{Author: "did:plc:owner", Source: domain.ListItemCollection, RecordKey: "member"}} + now := time.UnixMicro(1000) + if saved, err := legacy.SaveTrust(ctx, "path", trust.Result{Status: trust.CalculatedStatus, Score: &score}, deps, revision, now.Add(time.Hour)); !saved || err != nil { + t.Fatalf("cache: %v %v", saved, err) + } + if err := conn.Close(); err != nil { + t.Fatal(err) + } + s, err := OpenWithSoft(path, domain.SoftParticipantLists) + if err != nil { + t.Fatal(err) + } + defer s.Close() + rows, err := s.Incoming(ctx, follow.Subject, "", domain.Default, 100) + if err != nil || len(rows) != 2 { + t.Fatalf("migration lost sources: %+v %v", rows, err) + } + if _, err := s.CachedTrust(ctx, "path", now); err != nil { + t.Fatalf("migration lost cache: %v", err) + } + follow.RecordKey, follow.Subject = "deleted", "did:plc:stale" + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: follow.Author, Source: follow.Source, RecordKey: follow.RecordKey, Vouch: &follow, TimeUS: 200}); err != nil { + t.Fatal(err) + } + rows, err = s.Incoming(ctx, follow.Subject, "", domain.Default, 100) + if err != nil || len(rows) != 0 { + t.Fatalf("migration lost tombstone: %+v %v", rows, err) + } + if err := s.Apply(ctx, "stream", domain.Change{Kind: domain.RecordChange, DID: "did:plc:owner", Source: domain.ListItemCollection, RecordKey: "member", TimeUS: 400}); err != nil { + t.Fatal(err) + } + if _, err := s.CachedTrust(ctx, "path", now); err == nil { + t.Fatal("migration lost cache invalidation") + } + check, err := s.conn.Query("PRAGMA foreign_key_check") + if err != nil { + t.Fatal(err) + } + defer check.Close() + if check.Next() { + t.Fatal("migration broke foreign keys") + } +} diff --git a/internal/storage/lists.go b/internal/storage/lists.go index 0a5ad0e..4c4c65a 100644 --- a/internal/storage/lists.go +++ b/internal/storage/lists.go @@ -4,14 +4,15 @@ import ( "context" "database/sql" "errors" + "slices" "time" "provenance/internal/domain" "provenance/internal/storage/db" ) -func listRows(rows []db.ListEdge) []domain.Vouch { - result := make([]domain.Vouch, 0, len(rows)) +func appendListRows(result []domain.Vouch, rows []db.ListEdge) []domain.Vouch { + result = slices.Grow(result, len(rows)) for _, row := range rows { result = append(result, domain.Vouch{Author: row.Author, Subject: row.Subject, Score: domain.SoftBlockScore, Source: row.Source, RecordKey: row.RecordKey, CreatedAt: row.CreatedAt, ListURI: row.ListUri, ListAuthor: row.ListAuthor, ListKey: row.ListKey, ItemKey: row.ItemKey}) } diff --git a/internal/storage/lists_schema.sql b/internal/storage/lists_schema.sql index b0b0be0..5b524ed 100644 --- a/internal/storage/lists_schema.sql +++ b/internal/storage/lists_schema.sql @@ -7,7 +7,7 @@ CREATE TABLE IF NOT EXISTS list_records ( purpose TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, PRIMARY KEY(author, source, record_key) -); +) WITHOUT ROWID; CREATE INDEX IF NOT EXISTS list_records_uri ON list_records(list_uri, source, subject, author); CREATE INDEX IF NOT EXISTS list_records_subject ON list_records(subject, source, author); @@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS list_versions ( record_key TEXT NOT NULL, time_us INTEGER NOT NULL, PRIMARY KEY(author, source, record_key) -); +) WITHOUT ROWID; CREATE TABLE IF NOT EXISTS list_snapshots ( list_uri TEXT PRIMARY KEY, owner TEXT NOT NULL REFERENCES soft_accounts(did) ON DELETE CASCADE, diff --git a/internal/storage/migration7.sql b/internal/storage/migration7.sql new file mode 100644 index 0000000..34e0375 --- /dev/null +++ b/internal/storage/migration7.sql @@ -0,0 +1,20 @@ +DROP VIEW IF EXISTS soft_edges; +DROP VIEW IF EXISTS list_edges; +DROP TRIGGER IF EXISTS trust_soft_delete; +DROP TRIGGER IF EXISTS trust_soft_update; +DROP TRIGGER IF EXISTS graph_soft_insert; +DROP TRIGGER IF EXISTS graph_soft_delete; +DROP TRIGGER IF EXISTS graph_soft_update; +DROP TRIGGER IF EXISTS trust_list_delete; +DROP TRIGGER IF EXISTS trust_list_update; +DROP TRIGGER IF EXISTS graph_list_insert; +DROP TRIGGER IF EXISTS graph_list_delete; +DROP TRIGGER IF EXISTS graph_list_update; +DROP INDEX IF EXISTS soft_vouches_pair; +DROP INDEX IF EXISTS soft_vouches_subject; +DROP INDEX IF EXISTS list_records_uri; +DROP INDEX IF EXISTS list_records_subject; +ALTER TABLE soft_vouches RENAME TO legacy_soft_vouches; +ALTER TABLE soft_versions RENAME TO legacy_soft_versions; +ALTER TABLE list_records RENAME TO legacy_list_records; +ALTER TABLE list_versions RENAME TO legacy_list_versions; diff --git a/internal/storage/migration7_copy.sql b/internal/storage/migration7_copy.sql new file mode 100644 index 0000000..201457b --- /dev/null +++ b/internal/storage/migration7_copy.sql @@ -0,0 +1,8 @@ +INSERT INTO soft_vouches SELECT * FROM legacy_soft_vouches; +INSERT INTO soft_versions SELECT * FROM legacy_soft_versions; +INSERT INTO list_records SELECT * FROM legacy_list_records; +INSERT INTO list_versions SELECT * FROM legacy_list_versions; +DROP TABLE legacy_soft_vouches; +DROP TABLE legacy_soft_versions; +DROP TABLE legacy_list_records; +DROP TABLE legacy_list_versions; diff --git a/internal/storage/scale_test.go b/internal/storage/scale_test.go new file mode 100644 index 0000000..6cdac61 --- /dev/null +++ b/internal/storage/scale_test.go @@ -0,0 +1,72 @@ +package storage + +import ( + "context" + "fmt" + "path/filepath" + "testing" + + "provenance/internal/domain" + "provenance/internal/storage/db" + "provenance/internal/trust" +) + +// Include live replay versions and indexes so that bytes per record reflect ingestion storage. +func BenchmarkSoftScale(b *testing.B) { + for _, count := range []int{10000, 100000} { + b.Run(fmt.Sprint(count), func(b *testing.B) { + ctx := context.Background() + s, err := OpenWithSoft(filepath.Join(b.TempDir(), "scale.db"), domain.SoftAllAuthors) + if err != nil { + b.Fatal(err) + } + defer s.Close() + tx, err := s.conn.BeginTx(ctx, nil) + if err != nil { + b.Fatal(err) + } + q := s.queries.WithTx(tx) + const authorCount = 1000 + for index := 0; index < authorCount; index++ { + if err := q.JoinSoft(ctx, fmt.Sprintf("did:plc:%024d", index)); err != nil { + b.Fatal(err) + } + } + for index := 0; index < count; index++ { + author := fmt.Sprintf("did:plc:%024d", index%authorCount) + subject := fmt.Sprintf("did:plc:%024d", authorCount+index) + key := fmt.Sprintf("3%012d", index) + if err := q.ReplaceSoft(ctx, db.ReplaceSoftParams{Author: author, Subject: subject, Source: domain.FollowCollection, RecordKey: key, Score: domain.FollowScore, CreatedAt: "2026-10-09T00:00:00Z"}); err != nil { + b.Fatal(err) + } + if err := q.SetSoftVersion(ctx, db.SetSoftVersionParams{Author: author, Source: domain.FollowCollection, RecordKey: key, TimeUs: int64(index + 1)}); err != nil { + b.Fatal(err) + } + } + if err := tx.Commit(); err != nil { + b.Fatal(err) + } + var pages, pageSize int64 + if err := s.conn.QueryRow("PRAGMA page_count").Scan(&pages); err != nil { + b.Fatal(err) + } + if err := s.conn.QueryRow("PRAGMA page_size").Scan(&pageSize); err != nil { + b.Fatal(err) + } + bytesPerRecord := float64(pages*pageSize) / float64(count) + b.ReportAllocs() + b.ResetTimer() + for index := 0; index < b.N; index++ { + rows, _, err := s.TrustGraph(ctx, "", domain.Default, count+1) + if err != nil || len(rows) != count { + b.Fatalf("graph: %d %v", len(rows), err) + } + selected := trust.Select(rows, "", domain.Default) + if len(selected) != count { + b.Fatal("selection lost edges") + } + } + b.ReportMetric(bytesPerRecord, "disk-bytes/record") + }) + } +} diff --git a/internal/storage/soft.go b/internal/storage/soft.go index 43b7980..dcb3c12 100644 --- a/internal/storage/soft.go +++ b/internal/storage/soft.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "slices" "time" "provenance/internal/domain" @@ -30,16 +31,17 @@ func (s *SQLite) graphRows(ctx context.Context, q *db.Queries, tag string, mode if err != nil { return nil, err } - result = append(result, softRows(soft)...) + result = appendSoftRows(result, soft) if !domain.SoftLists(s.softMode) { return result, nil } lists, err := q.ListGraph(ctx, db.ListGraphParams{Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) - return append(result, listRows(lists)...), err + return appendListRows(result, lists), err } -func softRows(rows []db.SoftEdge) []domain.Vouch { - result := make([]domain.Vouch, 0, len(rows)) +func appendSoftRows(result []domain.Vouch, rows []db.SoftEdge) []domain.Vouch { + // Convert into the final buffer instead of a second full edge slice. + result = slices.Grow(result, len(rows)) for _, row := range rows { result = append(result, domain.Vouch{Author: row.Author, Subject: row.Subject, Score: row.Score, Source: row.Source, RecordKey: row.RecordKey, CreatedAt: row.CreatedAt}) } diff --git a/internal/storage/soft_schema.sql b/internal/storage/soft_schema.sql index 931418e..ff79b98 100644 --- a/internal/storage/soft_schema.sql +++ b/internal/storage/soft_schema.sql @@ -16,7 +16,7 @@ CREATE TABLE IF NOT EXISTS soft_vouches ( ), created_at TEXT NOT NULL, PRIMARY KEY(author, source, record_key) -); +) WITHOUT ROWID; CREATE INDEX IF NOT EXISTS soft_vouches_pair ON soft_vouches(author, subject, source, record_key); CREATE INDEX IF NOT EXISTS soft_vouches_subject ON soft_vouches(subject, author, source, record_key); @@ -39,7 +39,7 @@ CREATE TABLE IF NOT EXISTS soft_versions ( record_key TEXT NOT NULL, time_us INTEGER NOT NULL, PRIMARY KEY(author, source, record_key) -); +) WITHOUT ROWID; CREATE TRIGGER IF NOT EXISTS trust_soft_delete BEFORE DELETE ON soft_vouches BEGIN DELETE FROM trust_cache WHERE cache_key IN ( diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index fa3c6f5..ff8e9a8 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -37,7 +37,13 @@ var softSchema string //go:embed lists_schema.sql var listsSchema string -const SchemaVersion = 6 +//go:embed migration7.sql +var migration7 string + +//go:embed migration7_copy.sql +var migration7Copy string + +const SchemaVersion = 7 type SQLite struct { conn *sql.DB @@ -143,6 +149,20 @@ func migrate(conn *sql.DB) error { if _, err := tx.Exec(listsSchema); err != nil { return err } + if version > 0 { + if _, err := tx.Exec(migration7); err != nil { + return err + } + if _, err := tx.Exec(softSchema); err != nil { + return err + } + if _, err := tx.Exec(listsSchema); err != nil { + return err + } + if _, err := tx.Exec(migration7Copy); err != nil { + return err + } + } if _, err := tx.Exec(fmt.Sprintf("PRAGMA user_version=%d", SchemaVersion)); err != nil { return err } diff --git a/internal/storage/trust_cache.go b/internal/storage/trust_cache.go index e99d90c..d417c92 100644 --- a/internal/storage/trust_cache.go +++ b/internal/storage/trust_cache.go @@ -53,12 +53,12 @@ func (s *SQLite) Incoming(ctx context.Context, target, tag string, mode domain.M if err != nil { return nil, err } - result = append(result, softRows(soft)...) + result = appendSoftRows(result, soft) if !domain.SoftLists(s.softMode) { return result, nil } lists, err := s.queries.ListIncoming(ctx, db.ListIncomingParams{Subject: target, Tag: tag, Fallback: fallback, AllAuthors: all, MaxRows: int64(limit)}) - return append(result, listRows(lists)...), err + return appendListRows(result, lists), err } func (s *SQLite) CachedTrust(ctx context.Context, key string, now time.Time) (trust.Result, error) { diff --git a/internal/trust/graph.go b/internal/trust/graph.go index 85f2f5a..0398701 100644 --- a/internal/trust/graph.go +++ b/internal/trust/graph.go @@ -47,8 +47,9 @@ type Limits struct { // Select one assertion per pair so that fallback never overrides an explicit score. func Select(vouches []domain.Vouch, tag string, mode domain.Mode) []domain.Vouch { - chosen := make(map[[2]string]domain.Vouch) - for _, v := range vouches { + // Store input indexes instead of another full copy of every selected assertion. + chosen := make(map[[2]string]int, len(vouches)) + for index, v := range vouches { if domain.SoftSource(v.Source) && mode == domain.Tagged { continue } @@ -57,13 +58,13 @@ func Select(vouches []domain.Vouch, tag string, mode domain.Mode) []domain.Vouch } key := [2]string{v.Author, v.Subject} previous, exists := chosen[key] - if !exists || preferVouch(v, previous, tag) { - chosen[key] = v + if !exists || preferVouch(v, vouches[previous], tag) { + chosen[key] = index } } result := make([]domain.Vouch, 0, len(chosen)) - for _, v := range chosen { - result = append(result, v) + for _, index := range chosen { + result = append(result, vouches[index]) } sort.Slice(result, func(i, j int) bool { if result[i].Author != result[j].Author { diff --git a/internal/trust/scale_test.go b/internal/trust/scale_test.go new file mode 100644 index 0000000..e26e772 --- /dev/null +++ b/internal/trust/scale_test.go @@ -0,0 +1,22 @@ +package trust + +import ( + "fmt" + "provenance/internal/domain" + "testing" +) + +func BenchmarkSelectScale(b *testing.B) { + const count = 100000 + edges := make([]domain.Vouch, count) + for index := range edges { + edges[index] = domain.Vouch{Author: fmt.Sprintf("did:plc:%024d", index), Subject: "did:plc:target", Source: domain.FollowCollection, RecordKey: fmt.Sprint(index), Score: .5} + } + b.ReportAllocs() + b.ResetTimer() + for index := 0; index < b.N; index++ { + if len(Select(edges, "", domain.Default)) != count { + b.Fatal("selection lost edges") + } + } +}