denozord/EvoBGP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 131 Wiki Activity

feat(db): backfill prefix snapshots and switch read path

Browse Source 
Backfill миграция; чтение префиксов через snapshot; запись без дублирования rows.

Co-authored-by: Cursor <[email protected]>
This commit is contained in:
Denozordec
2026-05-25 10:58:01 +07:00
co-authored by Cursor
parent 8a9d60a5a7
commit f57b430052
7 changed files with 286 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/repository/postgres.go
+20 -38
View File
@@ -800,6 +800,9 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
} }
return nil, "", false return nil, "", false
} }
if snapID, ok := p.revisionPrefixSnapshotID(ctx, revisionID); ok {
return p.listSnapshotPrefixes(ctx, snapID, cursor, limit)
}
var rows pgx.Rows var rows pgx.Rows
var err error var err error
if useOffset { if useOffset {
@@ -884,11 +887,7 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st
if err := copyRevisionPreview(ctx, tx, newID, sourceRevisionID); err != nil { if err := copyRevisionPreview(ctx, tx, newID, sourceRevisionID); err != nil {
return "", err return "", err
} }
_, err = tx.Exec(ctx, ` if err := p.copyRevisionPrefixSnapshotRef(ctx, tx, newID, sourceRevisionID); err != nil {
INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source, meta_json)
SELECT $1::uuid, prefix, community_id, source, meta_json FROM revision_materialized_prefix WHERE revision_id=$2::uuid`,
newID, sourceRevisionID)
if err != nil {
return "", err return "", err
} }
if err := tx.Commit(ctx); err != nil { if err := tx.Commit(ctx); err != nil {
@@ -909,19 +908,17 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
ctx := context.Background() ctx := context.Background()
var unchanged int var unchanged int
err := p.pool.QueryRow(ctx, ` err := p.pool.QueryRow(ctx, `
SELECT COUNT(*)::int FROM revision_materialized_prefix b SELECT COUNT(*)::int FROM (
INNER JOIN revision_materialized_prefix a SELECT b.prefix FROM (`+sqlRevisionPrefixes("$2")+`) b
ON a.revision_id = $1::uuid AND a.prefix = b.prefix INNER JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
WHERE b.revision_id = $2::uuid`, aID, bID).Scan(&unchanged) ) t`, aID, bID).Scan(&unchanged)
if err != nil { if err != nil {
return nil, err return nil, err
} }
rowsAdded, err := p.pool.Query(ctx, ` rowsAdded, err := p.pool.Query(ctx, `
SELECT b.prefix::text SELECT b.prefix::text FROM (`+sqlRevisionPrefixes("$2")+`) b
FROM revision_materialized_prefix b LEFT JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
LEFT JOIN revision_materialized_prefix a WHERE a.prefix IS NULL
ON a.revision_id = $1::uuid AND a.prefix = b.prefix
WHERE b.revision_id = $2::uuid AND a.prefix IS NULL
ORDER BY b.prefix ORDER BY b.prefix
LIMIT $3`, aID, bID, maxRevisionDiffRows+1) LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
if err != nil { if err != nil {
@@ -942,11 +939,9 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
} }
addedTruncated := len(added) >= maxRevisionDiffRows addedTruncated := len(added) >= maxRevisionDiffRows
rowsRem, err := p.pool.Query(ctx, ` rowsRem, err := p.pool.Query(ctx, `
SELECT a.prefix::text SELECT a.prefix::text FROM (`+sqlRevisionPrefixes("$1")+`) a
FROM revision_materialized_prefix a LEFT JOIN (`+sqlRevisionPrefixes("$2")+`) b ON b.prefix = a.prefix
LEFT JOIN revision_materialized_prefix b WHERE b.prefix IS NULL
ON b.revision_id = $2::uuid AND b.prefix = a.prefix
WHERE a.revision_id = $1::uuid AND b.prefix IS NULL
ORDER BY a.prefix ORDER BY a.prefix
LIMIT $3`, aID, bID, maxRevisionDiffRows+1) LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
if err != nil { if err != nil {
@@ -1109,25 +1104,12 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
if err := insertRevisionPreview(ctx, tx, revID, previewFragments); err != nil { if err := insertRevisionPreview(ctx, tx, revID, previewFragments); err != nil {
return err return err
} }
if len(prefixes) > 0 { snapID, err := p.ensurePrefixSnapshot(ctx, tx, contentHash, prefixes)
_, err = tx.CopyFrom(ctx, if err != nil {
pgx.Identifier{"revision_materialized_prefix"}, return err
[]string{"revision_id", "prefix", "community_id", "source"}, }
pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) { if err := p.linkRevisionPrefixSnapshot(ctx, tx, revID, snapID); err != nil {
pr := prefixes[i] return err
var comm any
if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
comm = strings.TrimSpace(*pr.CommunityID)
}
src := pr.Source
if strings.TrimSpace(src) == "" {
src = "render"
}
return []any{revID, strings.TrimSpace(pr.Prefix), comm, src}, nil
}))
if err != nil {
return err
}
} }
if err := tx.Commit(ctx); err != nil { if err := tx.Commit(ctx); err != nil {
return err return err
internal/repository/postgres_prefix_snapshot.go
+186
View File
@@ -0,0 +1,186 @@
package repository
import (
"context"
"errors"
"strings"
"evobgp/internal/store"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
)
func prefixSnapshotTableExists(ctx context.Context, q queryRower) bool {
var n int
err := q.QueryRow(ctx, `
SELECT 1 FROM information_schema.tables
WHERE table_schema = 'public' AND table_name = 'prefix_snapshot'
LIMIT 1`).Scan(&n)
return err == nil
}
func normalizeSnapshotHash(contentHash string) string {
h := strings.TrimSpace(contentHash)
if strings.HasPrefix(h, "sha256:") {
h = strings.TrimPrefix(h, "sha256:")
}
if len(h) > 64 {
h = h[:64]
}
if len(h) < 64 {
h = h + strings.Repeat("0", 64-len(h))
}
return h
}
func (p *Postgres) revisionPrefixSnapshotID(ctx context.Context, revisionID string) (string, bool) {
if !prefixSnapshotTableExists(ctx, p.pool) {
return "", false
}
var snap *string
err := p.pool.QueryRow(ctx, `
SELECT prefix_snapshot_id::text FROM config_revision
WHERE id = $1::uuid AND prefix_snapshot_id IS NOT NULL`, revisionID).Scan(&snap)
if err != nil || snap == nil || strings.TrimSpace(*snap) == "" {
return "", false
}
return *snap, true
}
func (p *Postgres) ensurePrefixSnapshot(ctx context.Context, db execQuerier, contentHash string, prefixes []store.PrefixRow) (string, error) {
if !prefixSnapshotTableExists(ctx, db) {
return "", nil
}
hash := normalizeSnapshotHash(contentHash)
var existing string
err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&existing)
if err == nil && existing != "" {
return existing, nil
}
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
return "", err
}
snapID := uuid.NewString()
if _, err := db.Exec(ctx, `
INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)
ON CONFLICT (content_hash) DO NOTHING`, snapID, hash); err != nil {
return "", err
}
if err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&snapID); err != nil {
return "", err
}
var rowCount int
_ = db.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot_row WHERE snapshot_id = $1::uuid`, snapID).Scan(&rowCount)
if rowCount > 0 {
return snapID, nil
}
for i, pr := range prefixes {
var comm any
if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
comm = strings.TrimSpace(*pr.CommunityID)
}
src := pr.Source
if strings.TrimSpace(src) == "" {
src = "render"
}
if _, err := db.Exec(ctx, `
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
VALUES ($1::uuid, $2, $3::cidr, $4::uuid, $5)`,
snapID, i, strings.TrimSpace(pr.Prefix), comm, src); err != nil {
return "", err
}
}
return snapID, nil
}
func (p *Postgres) listSnapshotPrefixes(ctx context.Context, snapshotID, cursor string, limit int) ([]store.PrefixRow, string, bool) {
afterOrd, off, useOffset := store.ParsePrefixPageCursor(cursor)
var rows pgx.Rows
var err error
if useOffset {
rows, err = p.pool.Query(ctx, `
SELECT ord, prefix::text, community_id::text, source
FROM prefix_snapshot_row
WHERE snapshot_id = $1::uuid
ORDER BY ord
LIMIT $2 OFFSET $3`, snapshotID, limit+1, off)
} else {
var afterArg any
if afterOrd != nil {
afterArg = int(*afterOrd)
}
rows, err = p.pool.Query(ctx, `
SELECT ord, prefix::text, community_id::text, source
FROM prefix_snapshot_row
WHERE snapshot_id = $1::uuid AND ($2::int IS NULL OR ord > $2::int)
ORDER BY ord
LIMIT $3`, snapshotID, afterArg, limit+1)
}
if err != nil {
return nil, "", false
}
defer rows.Close()
var all []store.PrefixRow
var ords []int64
for rows.Next() {
var ord int
var pr store.PrefixRow
var comm *string
if err := rows.Scan(&ord, &pr.Prefix, &comm, &pr.Source); err != nil {
continue
}
pr.CommunityID = comm
ords = append(ords, int64(ord))
all = append(all, pr)
}
more := len(all) > limit
if more {
all = all[:limit]
ords = ords[:limit]
}
next := ""
if more && len(ords) > 0 {
next = store.FormatPrefixPageCursor(ords[len(ords)-1])
}
if len(all) == 0 {
return nil, "", false
}
return all, next, more
}
func (p *Postgres) linkRevisionPrefixSnapshot(ctx context.Context, db execQuerier, revisionID, snapshotID string) error {
if snapshotID == "" || !prefixSnapshotTableExists(ctx, db) {
return nil
}
_, err := db.Exec(ctx, `
UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`,
revisionID, snapshotID)
return err
}
func (p *Postgres) copyRevisionPrefixSnapshotRef(ctx context.Context, db execQuerier, dstRevisionID, srcRevisionID string) error {
if !prefixSnapshotTableExists(ctx, db) {
return nil
}
_, err := db.Exec(ctx, `
UPDATE config_revision dst
SET prefix_snapshot_id = src.prefix_snapshot_id
FROM config_revision src
WHERE dst.id = $1::uuid AND src.id = $2::uuid AND src.prefix_snapshot_id IS NOT NULL`,
dstRevisionID, srcRevisionID)
return err
}
func sqlRevisionPrefixes(revParam string) string {
return `SELECT psr.prefix FROM config_revision cr
JOIN prefix_snapshot_row psr ON psr.snapshot_id = cr.prefix_snapshot_id
WHERE cr.id = ` + revParam + `::uuid AND cr.prefix_snapshot_id IS NOT NULL
UNION ALL
SELECT rmp.prefix FROM revision_materialized_prefix rmp
WHERE rmp.revision_id = ` + revParam + `::uuid
AND NOT EXISTS (
SELECT 1 FROM config_revision cr2
WHERE cr2.id = ` + revParam + `::uuid AND cr2.prefix_snapshot_id IS NOT NULL
)`
}
internal/repository/postgres_seed.go
+13 -2
View File
@@ -85,9 +85,20 @@ protocol direct {
INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, rid, string(previewFragsB)); err != nil { INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, rid, string(previewFragsB)); err != nil {
return err return err
} }
snapID := uuid.NewString()
demoHash := normalizeSnapshotHash("sha256:demo-rev-1")
if _, err := tx.Exec(ctx, ` if _, err := tx.Exec(ctx, `
INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source) INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)`, snapID, demoHash); err != nil {
VALUES ($1::uuid,'203.0.113.0/24',$2::uuid,'demo'), ($1::uuid,'2001:db8::/32',$2::uuid,'demo')`, rid, cid); err != nil { return err
}
if _, err := tx.Exec(ctx, `
UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`, rid, snapID); err != nil {
return err
}
if _, err := tx.Exec(ctx, `
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
VALUES ($1::uuid, 0, '203.0.113.0/24'::cidr, $2::uuid, 'demo'),
($1::uuid, 1, '2001:db8::/32'::cidr, $2::uuid, 'demo')`, snapID, cid); err != nil {
return err return err
} }
if _, err := tx.Exec(ctx, ` if _, err := tx.Exec(ctx, `
migrations/postgres/000017_prefix_snapshot_backfill.down.sql
+3
View File
@@ -0,0 +1,3 @@
DELETE FROM prefix_snapshot_row;
UPDATE config_revision SET prefix_snapshot_id = NULL WHERE prefix_snapshot_id IS NOT NULL;
DELETE FROM prefix_snapshot;
migrations/postgres/000017_prefix_snapshot_backfill.up.sql
+31
View File
@@ -0,0 +1,31 @@
-- Backfill prefix snapshots from revision_materialized_prefix.
WITH new_snaps AS (
INSERT INTO prefix_snapshot (id, content_hash)
SELECT gen_random_uuid(),
substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64)
FROM config_revision cr
WHERE cr.prefix_snapshot_id IS NULL
AND EXISTS (
SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id
)
RETURNING id, content_hash
)
UPDATE config_revision cr
SET prefix_snapshot_id = ns.id
FROM new_snaps ns
WHERE cr.prefix_snapshot_id IS NULL
AND ns.content_hash = substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64);
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
SELECT cr.prefix_snapshot_id,
(row_number() OVER (PARTITION BY cr.id ORDER BY rmp.id) - 1)::int,
rmp.prefix,
rmp.community_id,
rmp.source
FROM config_revision cr
JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id
WHERE cr.prefix_snapshot_id IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id
);
migrations/sqlite/000017_prefix_snapshot_backfill.down.sql
+3
View File
@@ -0,0 +1,3 @@
DELETE FROM prefix_snapshot_row;
UPDATE config_revision SET prefix_snapshot_id = NULL;
DELETE FROM prefix_snapshot;
migrations/sqlite/000017_prefix_snapshot_backfill.up.sql
+30
View File
@@ -0,0 +1,30 @@
-- SQLite backfill: one snapshot per revision with materialized prefixes.
INSERT INTO prefix_snapshot (id, content_hash)
SELECT lower(hex(randomblob(16))),
substr(replace(cr.id, '-', '') || replace(cr.id, '-', ''), 1, 64)
FROM config_revision cr
WHERE cr.prefix_snapshot_id IS NULL
AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id);
UPDATE config_revision
SET prefix_snapshot_id = (
SELECT ps.id FROM prefix_snapshot ps
WHERE ps.content_hash = substr(replace(config_revision.id, '-', '') || replace(config_revision.id, '-', ''), 1, 64)
)
WHERE prefix_snapshot_id IS NULL
AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = config_revision.id);
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
SELECT cr.prefix_snapshot_id,
(SELECT COUNT(*) FROM revision_materialized_prefix r2
WHERE r2.revision_id = cr.id AND r2.id <= rmp.id) - 1,
rmp.prefix,
rmp.community_id,
rmp.source
FROM config_revision cr
JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id
WHERE cr.prefix_snapshot_id IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id
);