feat(revisions): add pruning estimate and cleanup endpoints
CI / changes (push) Successful in 8s
CI / commitlint (push) Has been skipped
CI / openapi (push) Successful in 25s
CI / web (push) Successful in 32s
CI / go (push) Successful in 57s
CI / bird2 (push) Successful in 15s
CI / release (push) Successful in 3m18s

Implemented new endpoints for estimating and pruning revisions, including detailed schemas for requests and responses. The `RevisionPruneEstimate` and `RevisionPruneResult` components were added to the OpenAPI documentation, enhancing the API's functionality for managing revision retention. Updated the backend to support these operations and integrated them into the tenant settings UI for improved user interaction.
This commit is contained in:
Denozordec
2026-06-12 21:56:36 +07:00
parent 5dbdac3d2c
commit f39df7c4bf
18 changed files with 1146 additions and 130 deletions
+124
View File
@@ -1212,6 +1212,70 @@ components:
format: date-time
additionalProperties: true
RevisionPruneEstimate:
type: object
required:
- retention_minutes
- cutoff_at
- revision_count
- prefix_row_count
- orphan_snapshot_count
- bytes_estimate
properties:
retention_minutes:
type: integer
minimum: 15
maximum: 43200
cutoff_at:
type: string
format: date-time
revision_count:
type: integer
minimum: 0
prefix_row_count:
type: integer
minimum: 0
description: Строки prefix_snapshot_row в освобождаемых снимках.
orphan_snapshot_count:
type: integer
minimum: 0
bytes_estimate:
type: integer
format: int64
minimum: 0
description: Ориентировочный логический объём данных (байты).
RevisionPruneResult:
type: object
required:
- deleted_revisions
- deleted_prefix_snapshots
- deleted_prefix_rows
- bytes_estimate
properties:
deleted_revisions:
type: integer
minimum: 0
deleted_prefix_snapshots:
type: integer
minimum: 0
deleted_prefix_rows:
type: integer
minimum: 0
bytes_estimate:
type: integer
format: int64
minimum: 0
RevisionPruneRequest:
type: object
properties:
retention_minutes:
type: integer
minimum: 15
maximum: 43200
description: TTL в минутах; если не задан — из revision_retention_minutes tenant settings.
PrefixSnapshotItem:
type: object
description: >
@@ -2823,6 +2887,66 @@ paths:
default:
$ref: "#/components/responses/DefaultProblem"
/v1/revisions/prune-estimate:
get:
tags: [Revisions]
summary: Оценка очистки ревизий по retention
description: >
Считает ревизии и ориентировочный объём данных, которые будут удалены при prune
(те же правила, что applyRevisionRetention: последняя ревизия tenant и раскатанные на спикерах сохраняются).
operationId: getRevisionPruneEstimate
parameters:
- $ref: "#/components/parameters/TenantId"
- name: retention_minutes
in: query
required: false
schema:
type: integer
minimum: 15
maximum: 43200
description: TTL в минутах; если не задан — из tenant settings (default 30d).
responses:
"200":
description: Оценка.
content:
application/json:
schema:
$ref: "#/components/schemas/RevisionPruneEstimate"
"422":
$ref: "#/components/responses/UnprocessableEntity"
default:
$ref: "#/components/responses/DefaultProblem"
/v1/revisions/prune:
post:
tags: [Revisions]
summary: Очистить старые ревизии (синхронно)
description: >
Удаляет ревизии старше cutoff по retention и GC неиспользуемых prefix_snapshot.
Operator-only.
operationId: pruneRevisions
parameters:
- $ref: "#/components/parameters/TenantId"
requestBody:
required: false
content:
application/json:
schema:
$ref: "#/components/schemas/RevisionPruneRequest"
responses:
"200":
description: Результат очистки.
content:
application/json:
schema:
$ref: "#/components/schemas/RevisionPruneResult"
"403":
$ref: "#/components/responses/Forbidden"
"422":
$ref: "#/components/responses/UnprocessableEntity"
default:
$ref: "#/components/responses/DefaultProblem"
/v1/revisions/{revision_id}:
parameters:
- $ref: "#/components/parameters/TenantId"
+2
View File
@@ -59,6 +59,8 @@ func (s *Server) registerV1(m *http.ServeMux) {
m.HandleFunc("POST /modules/{module_id}/refresh", s.handleModuleRefresh)
m.HandleFunc("POST /tenant/refresh", s.handleTenantRefresh)
m.HandleFunc("GET /revisions", s.handleListRevisions)
m.HandleFunc("GET /revisions/prune-estimate", s.handleRevisionPruneEstimate)
m.HandleFunc("POST /revisions/prune", s.handleRevisionPrune)
m.HandleFunc("GET /revisions/{revision_id}", s.handleGetRevision)
m.HandleFunc("GET /revisions/{revision_id}/preview", s.handleRevisionPreview)
m.HandleFunc("GET /revisions/{revision_id}/diagnostic-log", s.handleRevisionDiagnosticLog)
+101
View File
@@ -0,0 +1,101 @@
package httpapi
import (
"encoding/json"
"net/http"
"strconv"
"time"
"evobgp/internal/pipeline"
)
func (s *Server) resolveRevisionRetentionMinutesQuery(r *http.Request, tenantID string) (minutes int, ok bool) {
if raw := r.URL.Query().Get("retention_minutes"); raw != "" {
n, err := strconv.Atoi(raw)
if err != nil {
return 0, false
}
if n < pipeline.RevisionMinTTLMin || n > pipeline.RevisionMaxTTLMin {
return 0, false
}
return n, true
}
return s.defaultRevisionRetentionMinutes(tenantID)
}
func (s *Server) defaultRevisionRetentionMinutes(tenantID string) (int, bool) {
settings, err := s.store.ListGlobalSettings(tenantID)
if err != nil {
return pipeline.ClampRevisionRetentionMinutes(0), true
}
return pipeline.ClampRevisionRetentionMinutes(pipeline.RevisionRetentionMinutesFromSettings(settings)), true
}
func (s *Server) resolveRevisionRetentionMinutesBody(r *http.Request, tenantID string) (minutes int, ok bool) {
var body struct {
RetentionMinutes *int `json:"retention_minutes"`
}
if r.Body != nil && r.ContentLength != 0 {
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
return 0, false
}
if body.RetentionMinutes != nil {
m := *body.RetentionMinutes
if m < pipeline.RevisionMinTTLMin || m > pipeline.RevisionMaxTTLMin {
return 0, false
}
return m, true
}
}
return s.defaultRevisionRetentionMinutes(tenantID)
}
func (s *Server) handleRevisionPruneEstimate(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok || !s.requireAtLeast(w, a, "viewer") {
return
}
minutes, valid := s.resolveRevisionRetentionMinutesQuery(r, a.TenantID)
if !valid {
writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "retention_minutes must be an integer in range 15..43200")
return
}
cutoff := pipeline.RevisionCutoffFromMinutes(minutes)
est, err := s.store.EstimateRevisionPrune(a.TenantID, cutoff, minutes)
if err != nil {
writeInternalError(w, "revision prune estimate", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{
"retention_minutes": est.RetentionMinutes,
"cutoff_at": est.CutoffAt.UTC().Format(time.RFC3339Nano),
"revision_count": est.RevisionCount,
"prefix_row_count": est.PrefixRowCount,
"orphan_snapshot_count": est.OrphanSnapshotCount,
"bytes_estimate": est.BytesEstimate,
})
}
func (s *Server) handleRevisionPrune(w http.ResponseWriter, r *http.Request) {
a, ok := authFromContext(r.Context())
if !ok || !s.requireAtLeast(w, a, "operator") {
return
}
minutes, valid := s.resolveRevisionRetentionMinutesBody(r, a.TenantID)
if !valid {
writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "retention_minutes must be an integer in range 15..43200")
return
}
cutoff := pipeline.RevisionCutoffFromMinutes(minutes)
res, err := s.store.PruneRevisionsWithStats(a.TenantID, cutoff)
if err != nil {
writeInternalError(w, "revision prune", err)
return
}
writeJSON(w, http.StatusOK, map[string]any{
"deleted_revisions": res.DeletedRevisions,
"deleted_prefix_snapshots": res.DeletedPrefixSnapshots,
"deleted_prefix_rows": res.DeletedPrefixRows,
"bytes_estimate": res.BytesEstimate,
})
}
@@ -0,0 +1,94 @@
package httpapi
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
)
func TestRevisionPruneEstimateViewerOK(t *testing.T) {
srv, err := New(Options{InsecureDev: true, SeedDemo: true})
if err != nil {
t.Fatal(err)
}
defer srv.Close()
tenant, _, _, _, _ := srv.Store().DemoIDs()
mustSetTestAPIKeys(t, srv, "vwkey|"+tenant+"|viewer")
ts := httptest.NewServer(srv.Handler())
defer ts.Close()
req, _ := http.NewRequest(http.MethodGet, ts.URL+"/v1/revisions/prune-estimate?retention_minutes=15", nil)
req.Header.Set("Authorization", "Bearer vwkey")
resp, err := ts.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
t.Fatalf("status %d: %s", resp.StatusCode, b)
}
var body map[string]any
if err := json.NewDecoder(resp.Body).Decode(&body); err != nil {
t.Fatal(err)
}
if _, ok := body["revision_count"]; !ok {
t.Fatalf("missing revision_count: %v", body)
}
}
func TestRevisionPruneViewerForbidden(t *testing.T) {
srv, err := New(Options{InsecureDev: true, SeedDemo: true})
if err != nil {
t.Fatal(err)
}
defer srv.Close()
tenant, _, _, _, _ := srv.Store().DemoIDs()
mustSetTestAPIKeys(t, srv, "vwkey|"+tenant+"|viewer")
ts := httptest.NewServer(srv.Handler())
defer ts.Close()
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/revisions/prune", strings.NewReader(`{"retention_minutes":43200}`))
req.Header.Set("Authorization", "Bearer vwkey")
req.Header.Set("Content-Type", "application/json")
resp, err := ts.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusForbidden {
b, _ := io.ReadAll(resp.Body)
t.Fatalf("status %d: %s", resp.StatusCode, b)
}
}
func TestRevisionPruneOperatorOK(t *testing.T) {
srv, err := New(Options{InsecureDev: true, SeedDemo: true})
if err != nil {
t.Fatal(err)
}
defer srv.Close()
tenant, _, _, _, _ := srv.Store().DemoIDs()
mustSetTestAPIKeys(t, srv, "opkey|"+tenant+"|operator")
ts := httptest.NewServer(srv.Handler())
defer ts.Close()
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/revisions/prune", strings.NewReader(`{"retention_minutes":15}`))
req.Header.Set("Authorization", "Bearer opkey")
req.Header.Set("Content-Type", "application/json")
resp, err := ts.Client().Do(req)
if err != nil {
t.Fatal(err)
}
defer func() { _ = resp.Body.Close() }()
if resp.StatusCode != http.StatusOK {
b, _ := io.ReadAll(resp.Body)
t.Fatalf("status %d: %s", resp.StatusCode, b)
}
}
+2 -15
View File
@@ -31,10 +31,7 @@ const (
birdFilterNameV4 = "evobgp_export_v4"
birdFilterNameV6 = "evobgp_export_v6"
auxBirdFullExpanded = "_bird_full_expanded.conf"
revisionTTLKey = "revision_retention_minutes"
revisionMinTTLMin = 15
revisionMaxTTLMin = 30 * 24 * 60
revisionDefaultTTL = 30 * 24 * time.Hour
revisionTTLKey = RevisionRetentionKey
)
// AuxBirdFullExpandedKey returns the preview map key for the expanded BIRD config (generated on demand).
@@ -969,17 +966,7 @@ func applyRevisionRetention(st store.Backend, tenantID string) {
if err != nil {
return
}
ttl := revisionDefaultTTL
if minutes := intFromSettingsMap(settings, revisionTTLKey); minutes > 0 {
if minutes < revisionMinTTLMin {
minutes = revisionMinTTLMin
}
if minutes > revisionMaxTTLMin {
minutes = revisionMaxTTLMin
}
ttl = time.Duration(minutes) * time.Minute
}
cutoff := time.Now().UTC().Add(-ttl)
cutoff := RevisionRetentionCutoff(settings)
_, _ = st.PruneRevisionsBefore(tenantID, cutoff)
}
+75
View File
@@ -0,0 +1,75 @@
package pipeline
import (
"strconv"
"strings"
"time"
)
const (
// RevisionRetentionKey is the global_settings KV for revision TTL (minutes).
RevisionRetentionKey = "revision_retention_minutes"
RevisionMinTTLMin = 15
RevisionMaxTTLMin = 30 * 24 * 60
RevisionDefaultTTL = 30 * 24 * time.Hour
)
// RevisionCutoffFromMinutes returns created_at cutoff for revision prune (UTC now minus clamped TTL).
func RevisionCutoffFromMinutes(minutes int) time.Time {
ttl := RevisionDefaultTTL
if minutes > 0 {
m := minutes
if m < RevisionMinTTLMin {
m = RevisionMinTTLMin
}
if m > RevisionMaxTTLMin {
m = RevisionMaxTTLMin
}
ttl = time.Duration(m) * time.Minute
}
return time.Now().UTC().Add(-ttl)
}
// RevisionRetentionMinutesFromSettings reads revision_retention_minutes from tenant KV (0 if unset).
func RevisionRetentionMinutesFromSettings(settings map[string]any) int {
if settings == nil {
return 0
}
v, ok := settings[RevisionRetentionKey]
if !ok || v == nil {
return 0
}
switch x := v.(type) {
case float64:
return int(x)
case int:
return x
case int64:
return int(x)
case string:
n, err := strconv.Atoi(strings.TrimSpace(x))
if err == nil {
return n
}
}
return 0
}
// RevisionRetentionCutoff resolves tenant revision_retention_minutes from settings (default 30d).
func RevisionRetentionCutoff(settings map[string]any) time.Time {
return RevisionCutoffFromMinutes(RevisionRetentionMinutesFromSettings(settings))
}
// ClampRevisionRetentionMinutes normalizes user input to the allowed range.
func ClampRevisionRetentionMinutes(minutes int) int {
if minutes <= 0 {
return int(RevisionDefaultTTL / time.Minute)
}
if minutes < RevisionMinTTLMin {
return RevisionMinTTLMin
}
if minutes > RevisionMaxTTLMin {
return RevisionMaxTTLMin
}
return minutes
}
@@ -0,0 +1,30 @@
package pipeline
import (
"testing"
"time"
)
func TestClampRevisionRetentionMinutes(t *testing.T) {
if got := ClampRevisionRetentionMinutes(0); got != int(RevisionDefaultTTL/time.Minute) {
t.Fatalf("default: got %d", got)
}
if got := ClampRevisionRetentionMinutes(5); got != RevisionMinTTLMin {
t.Fatalf("min clamp: got %d", got)
}
if got := ClampRevisionRetentionMinutes(999999); got != RevisionMaxTTLMin {
t.Fatalf("max clamp: got %d", got)
}
if got := ClampRevisionRetentionMinutes(120); got != 120 {
t.Fatalf("unchanged: got %d", got)
}
}
func TestRevisionCutoffFromMinutes(t *testing.T) {
before := time.Now().UTC()
cutoff := RevisionCutoffFromMinutes(60)
after := time.Now().UTC().Add(-59 * time.Minute)
if cutoff.After(before.Add(-59*time.Minute)) || cutoff.Before(after.Add(-2*time.Minute)) {
t.Fatalf("cutoff out of range: %v", cutoff)
}
}
-37
View File
@@ -970,43 +970,6 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
}, nil
}
func (p *Postgres) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
ctx := context.Background()
total := 0
const batchSize = 50
for {
cmd, err := p.pool.Exec(ctx, `
DELETE FROM config_revision AS cr
WHERE cr.id IN (
SELECT id FROM config_revision
WHERE tenant_id = $1
AND created_at < $2
AND id <> (
SELECT id FROM config_revision
WHERE tenant_id = $1
ORDER BY created_at DESC
LIMIT 1
)
AND NOT EXISTS (
SELECT 1 FROM bgp_speaker AS sp
WHERE sp.tenant_id = $1
AND (sp.last_applied_revision_id = config_revision.id OR sp.published_revision_id = config_revision.id)
)
ORDER BY created_at ASC
LIMIT $3
)`, tenantID, cutoff.UTC(), batchSize)
if err != nil {
return total, err
}
n := int(cmd.RowsAffected())
total += n
if n < batchSize {
break
}
}
return total, nil
}
func (p *Postgres) SetLastAppliedRevision(tenantID, speakerID, revisionID string) error {
ctx := context.Background()
tag, err := p.pool.Exec(ctx, `
+23 -15
View File
@@ -53,21 +53,8 @@ func (p *Postgres) ensurePrefixSnapshot(ctx context.Context, db execQuerier, con
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 {
snapID, err := p.resolvePrefixSnapshotID(ctx, db, hash)
if err != nil {
return "", err
}
var rowCount int
@@ -94,6 +81,27 @@ func (p *Postgres) ensurePrefixSnapshot(ctx context.Context, db execQuerier, con
return snapID, nil
}
func (p *Postgres) resolvePrefixSnapshotID(ctx context.Context, db execQuerier, hash string) (string, error) {
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
}
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
@@ -0,0 +1,119 @@
package repository
import (
"context"
"os"
"testing"
"evobgp/internal/db"
"evobgp/internal/store"
"github.com/google/uuid"
)
func TestEnsurePrefixSnapshotFillsEmptyExistingIntegration(t *testing.T) {
dsn := os.Getenv("EVOBGP_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("EVOBGP_TEST_DATABASE_URL not set")
}
ctx := context.Background()
pool, err := db.OpenPostgresPool(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
pg, err := NewPostgres(ctx, pool, false)
if err != nil {
t.Fatal(err)
}
hash := normalizeSnapshotHash("sha256:empty-fill-" + uuid.NewString())
snapID := uuid.NewString()
if _, err := pool.Exec(ctx, `INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)`, snapID, hash); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, `DELETE FROM prefix_snapshot WHERE id = $1::uuid`, snapID)
})
tx, err := pool.Begin(ctx)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx.Rollback(ctx) }()
got, err := pg.ensurePrefixSnapshot(ctx, tx, "sha256:"+hash, []store.PrefixRow{
{Prefix: "203.0.113.1/32", Source: "test"},
})
if err != nil {
t.Fatal(err)
}
if got != snapID {
t.Fatalf("snap id: got %q want %q", got, snapID)
}
var rowCount int
if err := tx.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot_row WHERE snapshot_id = $1::uuid`, snapID).Scan(&rowCount); err != nil {
t.Fatal(err)
}
if rowCount != 1 {
t.Fatalf("row count: got %d", rowCount)
}
if err := tx.Commit(ctx); err != nil {
t.Fatal(err)
}
}
func TestEnsurePrefixSnapshotIdempotentIntegration(t *testing.T) {
dsn := os.Getenv("EVOBGP_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("EVOBGP_TEST_DATABASE_URL not set")
}
ctx := context.Background()
pool, err := db.OpenPostgresPool(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
pg, err := NewPostgres(ctx, pool, false)
if err != nil {
t.Fatal(err)
}
contentHash := "sha256:idempotent-" + uuid.NewString()
prefixes := []store.PrefixRow{{Prefix: "198.51.100.0/24", Source: "test"}}
tx1, err := pool.Begin(ctx)
if err != nil {
t.Fatal(err)
}
id1, err := pg.ensurePrefixSnapshot(ctx, tx1, contentHash, prefixes)
if err != nil {
t.Fatal(err)
}
if err := tx1.Commit(ctx); err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_, _ = pool.Exec(ctx, `DELETE FROM prefix_snapshot WHERE id = $1::uuid`, id1)
})
tx2, err := pool.Begin(ctx)
if err != nil {
t.Fatal(err)
}
defer func() { _ = tx2.Rollback(ctx) }()
id2, err := pg.ensurePrefixSnapshot(ctx, tx2, contentHash, prefixes)
if err != nil {
t.Fatal(err)
}
if id1 != id2 {
t.Fatalf("ids differ: %q vs %q", id1, id2)
}
var rowCount int
if err := tx2.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot_row WHERE snapshot_id = $1::uuid`, id1).Scan(&rowCount); err != nil {
t.Fatal(err)
}
if rowCount != 1 {
t.Fatalf("expected 1 row, got %d", rowCount)
}
}
@@ -0,0 +1,176 @@
package repository
import (
"context"
"time"
"evobgp/internal/store"
)
const revisionPruneBatchSize = 50
// sqlPrunableRevisionsWhere appends prunable revision predicates (tenant + cutoff params).
func sqlPrunableRevisionsWhere(tenantParam, cutoffParam string) string {
return `tenant_id = ` + tenantParam + `
AND created_at < ` + cutoffParam + `
AND id <> (
SELECT id FROM config_revision
WHERE tenant_id = ` + tenantParam + `
ORDER BY created_at DESC
LIMIT 1
)
AND NOT EXISTS (
SELECT 1 FROM bgp_speaker AS sp
WHERE sp.tenant_id = ` + tenantParam + `
AND (sp.last_applied_revision_id = config_revision.id OR sp.published_revision_id = config_revision.id)
)`
}
func (p *Postgres) EstimateRevisionPrune(tenantID string, cutoff time.Time, retentionMinutes int) (store.RevisionPruneEstimate, error) {
ctx := context.Background()
out := store.RevisionPruneEstimate{
RetentionMinutes: retentionMinutes,
CutoffAt: cutoff.UTC(),
}
where := sqlPrunableRevisionsWhere("$1", "$2")
var revBytes, previewBytes, rmpBytes, snapRowBytes int64
err := p.pool.QueryRow(ctx, `
WITH prunable AS (
SELECT id, prefix_snapshot_id, content_hash, meta_json
FROM config_revision
WHERE `+where+`
),
freed_snaps AS (
SELECT DISTINCT pr.prefix_snapshot_id AS snap_id
FROM prunable pr
WHERE pr.prefix_snapshot_id IS NOT NULL
EXCEPT
SELECT DISTINCT cr.prefix_snapshot_id
FROM config_revision cr
WHERE cr.prefix_snapshot_id IS NOT NULL
AND cr.id NOT IN (SELECT id FROM prunable)
)
SELECT
(SELECT COUNT(*)::int FROM prunable),
COALESCE((SELECT SUM(octet_length(content_hash) + octet_length(meta_json::text))::bigint FROM prunable), 0),
COALESCE((
SELECT SUM(octet_length(value))::bigint
FROM config_revision_preview p
JOIN prunable pr ON pr.id = p.revision_id
CROSS JOIN LATERAL jsonb_each_text(COALESCE(p.fragments, '{}'::jsonb))
), 0),
COALESCE((
SELECT SUM(octet_length(rmp.prefix) + octet_length(COALESCE(rmp.source, '')))::bigint
FROM revision_materialized_prefix rmp
WHERE rmp.revision_id IN (SELECT id FROM prunable)
), 0),
(SELECT COUNT(*)::int FROM freed_snaps),
COALESCE((SELECT COUNT(*)::int FROM prefix_snapshot_row psr WHERE psr.snapshot_id IN (SELECT snap_id FROM freed_snaps)), 0),
COALESCE((
SELECT SUM(octet_length(psr.prefix) + octet_length(COALESCE(psr.source, '')))::bigint
FROM prefix_snapshot_row psr
WHERE psr.snapshot_id IN (SELECT snap_id FROM freed_snaps)
), 0)`,
tenantID, cutoff.UTC()).Scan(
&out.RevisionCount,
&revBytes,
&previewBytes,
&rmpBytes,
&out.OrphanSnapshotCount,
&out.PrefixRowCount,
&snapRowBytes,
)
if err != nil {
return out, err
}
out.BytesEstimate = revBytes + previewBytes + rmpBytes + snapRowBytes
return out, nil
}
func (p *Postgres) PruneRevisionsWithStats(tenantID string, cutoff time.Time) (store.RevisionPruneResult, error) {
ctx := context.Background()
est, err := p.EstimateRevisionPrune(tenantID, cutoff, 0)
if err != nil {
return store.RevisionPruneResult{}, err
}
out := store.RevisionPruneResult{BytesEstimate: est.BytesEstimate}
where := sqlPrunableRevisionsWhere("$1", "$2")
for {
cmd, err := p.pool.Exec(ctx, `
DELETE FROM config_revision AS cr
WHERE cr.id IN (
SELECT id FROM config_revision
WHERE `+where+`
ORDER BY created_at ASC
LIMIT $3
)`, tenantID, cutoff.UTC(), revisionPruneBatchSize)
if err != nil {
return out, err
}
n := int(cmd.RowsAffected())
out.DeletedRevisions += n
if n < revisionPruneBatchSize {
break
}
}
for {
snaps, rows, err := p.pruneUnreferencedPrefixSnapshots(ctx, revisionPruneBatchSize)
if err != nil {
return out, err
}
out.DeletedPrefixSnapshots += snaps
out.DeletedPrefixRows += rows
if snaps < revisionPruneBatchSize {
break
}
}
return out, nil
}
func (p *Postgres) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
res, err := p.PruneRevisionsWithStats(tenantID, cutoff)
if err != nil {
return 0, err
}
return res.DeletedRevisions, nil
}
func (p *Postgres) pruneUnreferencedPrefixSnapshots(ctx context.Context, batchSize int) (deletedSnapshots int, deletedRows int, err error) {
if !prefixSnapshotTableExists(ctx, p.pool) {
return 0, 0, nil
}
var snapCount, rowCount int
err = p.pool.QueryRow(ctx, `
WITH doomed AS (
SELECT ps.id FROM prefix_snapshot ps
WHERE NOT EXISTS (
SELECT 1 FROM config_revision cr WHERE cr.prefix_snapshot_id = ps.id
)
LIMIT $1
)
SELECT
(SELECT COUNT(*)::int FROM doomed),
(SELECT COUNT(*)::int FROM prefix_snapshot_row psr WHERE psr.snapshot_id IN (SELECT id FROM doomed))`,
batchSize).Scan(&snapCount, &rowCount)
if err != nil {
return 0, 0, err
}
if snapCount == 0 {
return 0, 0, nil
}
_, err = p.pool.Exec(ctx, `
DELETE FROM prefix_snapshot
WHERE id IN (
SELECT ps.id FROM prefix_snapshot ps
WHERE NOT EXISTS (
SELECT 1 FROM config_revision cr WHERE cr.prefix_snapshot_id = ps.id
)
LIMIT $1
)`, batchSize)
if err != nil {
return 0, 0, err
}
return snapCount, rowCount, nil
}
@@ -0,0 +1,83 @@
package repository
import (
"context"
"os"
"testing"
"evobgp/internal/db"
"evobgp/internal/pipeline"
"github.com/google/uuid"
)
func TestPostgresPruneUnreferencedPrefixSnapshotsIntegration(t *testing.T) {
dsn := os.Getenv("EVOBGP_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("EVOBGP_TEST_DATABASE_URL not set")
}
ctx := context.Background()
pool, err := db.OpenPostgresPool(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
pg, err := NewPostgres(ctx, pool, false)
if err != nil {
t.Fatal(err)
}
snapID := uuid.NewString()
hash := normalizeSnapshotHash("sha256:test-orphan-" + snapID)
if _, err := pool.Exec(ctx, `INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)`, snapID, hash); err != nil {
t.Fatal(err)
}
if _, err := pool.Exec(ctx, `
INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, source)
VALUES ($1::uuid, 0, '203.0.113.0/24', 'test')`, snapID); err != nil {
t.Fatal(err)
}
snaps, rows, err := pg.pruneUnreferencedPrefixSnapshots(ctx, 50)
if err != nil {
t.Fatal(err)
}
if snaps < 1 || rows < 1 {
t.Fatalf("expected orphan cleanup, got snaps=%d rows=%d", snaps, rows)
}
var n int
if err := pool.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot WHERE id = $1::uuid`, snapID).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Fatalf("snapshot still exists")
}
}
func TestPostgresEstimateRevisionPruneIntegration(t *testing.T) {
dsn := os.Getenv("EVOBGP_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("EVOBGP_TEST_DATABASE_URL not set")
}
ctx := context.Background()
pool, err := db.OpenPostgresPool(ctx, dsn)
if err != nil {
t.Fatal(err)
}
defer pool.Close()
pg, err := NewPostgres(ctx, pool, false)
if err != nil {
t.Fatal(err)
}
tenant := uuid.NewString()
cutoff := pipeline.RevisionCutoffFromMinutes(15)
est, err := pg.EstimateRevisionPrune(tenant, cutoff, 15)
if err != nil {
t.Fatal(err)
}
if est.RevisionCount != 0 {
t.Fatalf("expected 0 revisions for empty tenant, got %d", est.RevisionCount)
}
}
+2
View File
@@ -84,6 +84,8 @@ type Backend interface {
// CreateRenderRevision inserts a new config_revision (revID must be unique) with materialized prefixes and preview fragments.
CreateRenderRevision(revisionID, tenantID, moduleID string, parentRevisionID *string, contentHash string, previewFragments map[string]string, prefixes []PrefixRow) error
RevisionDiff(tenantID, aID, bID string) (map[string]any, error)
EstimateRevisionPrune(tenantID string, cutoff time.Time, retentionMinutes int) (RevisionPruneEstimate, error)
PruneRevisionsWithStats(tenantID string, cutoff time.Time) (RevisionPruneResult, error)
PruneRevisionsBefore(tenantID string, cutoff time.Time) (deleted int, err error)
SetLastAppliedRevision(tenantID, speakerID, revisionID string) error
PublishRevisionForSpeaker(speakerID, revisionID string) error
-56
View File
@@ -616,62 +616,6 @@ func (m *Memory) RevisionDiff(tenantID, aID, bID string) (map[string]any, error)
}, nil
}
func (m *Memory) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
m.mu.Lock()
defer m.mu.Unlock()
var newest *Revision
for _, rev := range m.revisions {
if rev.TenantID != tenantID {
continue
}
if newest == nil || rev.CreatedAt.After(newest.CreatedAt) {
newest = rev
}
}
if newest == nil {
return 0, nil
}
protected := map[string]struct{}{
newest.ID: {},
}
for _, sp := range m.speakers {
if sp == nil || sp.TenantID != tenantID {
continue
}
if sp.LastAppliedRevisionID != nil && strings.TrimSpace(*sp.LastAppliedRevisionID) != "" {
protected[strings.TrimSpace(*sp.LastAppliedRevisionID)] = struct{}{}
}
}
for speakerID, info := range m.publishedRevision {
sp, ok := m.speakers[speakerID]
if !ok || sp == nil || sp.TenantID != tenantID {
continue
}
if strings.TrimSpace(info.RevisionID) != "" {
protected[strings.TrimSpace(info.RevisionID)] = struct{}{}
}
}
deleted := 0
for id, rev := range m.revisions {
if rev == nil || rev.TenantID != tenantID {
continue
}
if !rev.CreatedAt.Before(cutoff) {
continue
}
if _, keep := protected[id]; keep {
continue
}
delete(m.revisions, id)
delete(m.revPrefixes, id)
deleted++
}
return deleted, nil
}
func (m *Memory) getRevisionLocked(tenantID, revisionID string) (*Revision, error) {
rev, ok := m.revisions[revisionID]
if !ok {
+118
View File
@@ -0,0 +1,118 @@
package store
import (
"strings"
"time"
)
func (m *Memory) prunableRevisionIDsLocked(tenantID string, cutoff time.Time) []string {
var newest *Revision
for _, rev := range m.revisions {
if rev == nil || rev.TenantID != tenantID {
continue
}
if newest == nil || rev.CreatedAt.After(newest.CreatedAt) {
newest = rev
}
}
if newest == nil {
return nil
}
protected := map[string]struct{}{newest.ID: {}}
for _, sp := range m.speakers {
if sp == nil || sp.TenantID != tenantID {
continue
}
if sp.LastAppliedRevisionID != nil && strings.TrimSpace(*sp.LastAppliedRevisionID) != "" {
protected[strings.TrimSpace(*sp.LastAppliedRevisionID)] = struct{}{}
}
}
for speakerID, info := range m.publishedRevision {
sp, ok := m.speakers[speakerID]
if !ok || sp == nil || sp.TenantID != tenantID {
continue
}
if strings.TrimSpace(info.RevisionID) != "" {
protected[strings.TrimSpace(info.RevisionID)] = struct{}{}
}
}
var ids []string
for id, rev := range m.revisions {
if rev == nil || rev.TenantID != tenantID {
continue
}
if !rev.CreatedAt.Before(cutoff) {
continue
}
if _, keep := protected[id]; keep {
continue
}
ids = append(ids, id)
}
return ids
}
func revisionBytesEstimate(rev *Revision, prefixes []PrefixRow) int64 {
if rev == nil {
return 0
}
var n int64
n += int64(len(rev.ContentHash))
for _, v := range rev.PreviewFragments {
n += int64(len(v))
}
for _, pr := range prefixes {
n += int64(len(pr.Prefix) + len(pr.Source))
}
return n
}
func (m *Memory) EstimateRevisionPrune(tenantID string, cutoff time.Time, retentionMinutes int) (RevisionPruneEstimate, error) {
m.mu.Lock()
defer m.mu.Unlock()
ids := m.prunableRevisionIDsLocked(tenantID, cutoff)
var bytes int64
for _, id := range ids {
bytes += revisionBytesEstimate(m.revisions[id], m.revPrefixes[id])
}
prefixRows := 0
for _, id := range ids {
prefixRows += len(m.revPrefixes[id])
}
return RevisionPruneEstimate{
RetentionMinutes: retentionMinutes,
CutoffAt: cutoff.UTC(),
RevisionCount: len(ids),
PrefixRowCount: prefixRows,
OrphanSnapshotCount: 0,
BytesEstimate: bytes,
}, nil
}
func (m *Memory) PruneRevisionsWithStats(tenantID string, cutoff time.Time) (RevisionPruneResult, error) {
est, err := m.EstimateRevisionPrune(tenantID, cutoff, 0)
if err != nil {
return RevisionPruneResult{}, err
}
m.mu.Lock()
defer m.mu.Unlock()
ids := m.prunableRevisionIDsLocked(tenantID, cutoff)
for _, id := range ids {
delete(m.revisions, id)
delete(m.revPrefixes, id)
}
return RevisionPruneResult{
DeletedRevisions: len(ids),
DeletedPrefixSnapshots: 0,
DeletedPrefixRows: est.PrefixRowCount,
BytesEstimate: est.BytesEstimate,
}, nil
}
func (m *Memory) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
res, err := m.PruneRevisionsWithStats(tenantID, cutoff)
if err != nil {
return 0, err
}
return res.DeletedRevisions, nil
}
+21
View File
@@ -0,0 +1,21 @@
package store
import "time"
// RevisionPruneEstimate describes revisions and storage that would be removed by prune.
type RevisionPruneEstimate struct {
RetentionMinutes int `json:"retention_minutes"`
CutoffAt time.Time `json:"cutoff_at"`
RevisionCount int `json:"revision_count"`
PrefixRowCount int `json:"prefix_row_count"`
OrphanSnapshotCount int `json:"orphan_snapshot_count"`
BytesEstimate int64 `json:"bytes_estimate"`
}
// RevisionPruneResult is the outcome of a revision prune run.
type RevisionPruneResult struct {
DeletedRevisions int `json:"deleted_revisions"`
DeletedPrefixSnapshots int `json:"deleted_prefix_snapshots"`
DeletedPrefixRows int `json:"deleted_prefix_rows"`
BytesEstimate int64 `json:"bytes_estimate"`
}
@@ -2,10 +2,17 @@
import { onMount } from 'svelte';
import { defaults, superForm } from 'sveltekit-superforms';
import { zod4 } from 'sveltekit-superforms/adapters';
import type { AuthSession } from '$lib/api/types.js';
import { apiJSON } from '$lib/api/client.js';
import {
emptyRevisionSettingsForm,
revisionSettingsSchema
} from '$lib/settings/revision-settings.schema.js';
import {
fetchRevisionPruneEstimate,
pruneRevisionsNow,
type RevisionPruneEstimate
} from '$lib/settings/revision-prune-api.js';
import {
buildPayloadFromFormFields,
loadSettings,
@@ -13,6 +20,7 @@
patchSettings
} from '$lib/settings/settings-api.js';
import { REVISION_SETTING_KEYS } from '$lib/settings/settings-known-keys.js';
import { formatBytes } from '$lib/monitoring/postgres.js';
import { Button } from '$lib/ui/core/button/index.js';
import {
Card,
@@ -23,12 +31,18 @@
} from '$lib/ui/core/card/index.js';
import { Input } from '$lib/ui/core/input/index.js';
import FormField from '$lib/ui/patterns/form/form-field.svelte';
import { confirm } from '$lib/ui/patterns/confirm/confirm-state.svelte.js';
import { notify, notifyApiError } from '$lib/ui/app/toast.js';
import Save from '@lucide/svelte/icons/save';
import Trash2 from '@lucide/svelte/icons/trash-2';
let loading = $state(false);
let saving = $state(false);
let pruning = $state(false);
let loaded = $state(false);
let session = $state<AuthSession | null>(null);
let estimateLoading = $state(false);
let estimate = $state<RevisionPruneEstimate | null>(null);
const { form, errors, reset, validateForm } = superForm(
defaults(emptyRevisionSettingsForm(), zod4(revisionSettingsSchema)),
@@ -39,13 +53,60 @@
}
);
let hasValidationErrors = $derived(Boolean($errors.revision_retention_minutes?.length));
const isOperator = $derived(session?.role === 'operator');
const hasValidationErrors = $derived(Boolean($errors.revision_retention_minutes?.length));
let canSave = $derived.by(() => {
const parsedRetentionMinutes = $derived.by(() => {
const s = String($form.revision_retention_minutes ?? '').trim();
if (s === '' || !/^\d+$/.test(s)) return null;
const n = Number(s);
if (!Number.isInteger(n) || n < 15 || n > 43200) return null;
return n;
});
const canSave = $derived.by(() => {
if (loading || saving || hasValidationErrors || !loaded) return false;
return String($form.revision_retention_minutes ?? '').trim() !== '';
});
const canPruneNow = $derived.by(() => {
if (!isOperator || !loaded || pruning || saving || parsedRetentionMinutes === null)
return false;
return (estimate?.revision_count ?? 0) > 0;
});
async function loadSession() {
try {
session = await apiJSON<AuthSession>('/v1/auth/session');
} catch {
session = null;
}
}
async function refreshEstimate(minutes: number) {
estimateLoading = true;
try {
estimate = await fetchRevisionPruneEstimate(minutes);
} catch (e) {
estimate = null;
notifyApiError(e, 'Не удалось рассчитать оценку очистки');
} finally {
estimateLoading = false;
}
}
$effect(() => {
const minutes = parsedRetentionMinutes;
if (!loaded || minutes === null) {
estimate = null;
return;
}
const handle = setTimeout(() => {
void refreshEstimate(minutes);
}, 400);
return () => clearTimeout(handle);
});
async function load() {
loading = true;
try {
@@ -89,7 +150,46 @@
}
}
async function requestPruneNow() {
const minutes = parsedRetentionMinutes;
if (minutes === null) return;
let est: RevisionPruneEstimate;
try {
est = await fetchRevisionPruneEstimate(minutes);
} catch (e) {
notifyApiError(e);
return;
}
estimate = est;
if (est.revision_count === 0) {
notify.info('Нет ревизий для удаления по выбранному retention');
return;
}
void confirm({
title: 'Очистить старые ревизии?',
description: `Будет удалено ${est.revision_count} ревизий. Ориентировочно освободится ~${formatBytes(est.bytes_estimate)}. Действие необратимо.`,
confirmLabel: 'Очистить',
destructive: true,
onConfirm: async () => {
pruning = true;
try {
const res = await pruneRevisionsNow(minutes);
notify.success(
`Удалено ревизий: ${res.deleted_revisions}, освобождено ~${formatBytes(res.bytes_estimate)}`
);
await refreshEstimate(minutes);
} catch (e) {
notifyApiError(e);
throw e;
} finally {
pruning = false;
}
}
});
}
onMount(() => {
void loadSession();
void load();
});
</script>
@@ -98,7 +198,8 @@
<CardHeader>
<CardTitle>Хранение ревизий</CardTitle>
<CardDescription>
Автоматическая очистка старых ревизий. Последняя раскатанная ревизия не удаляется.
Автоматическая очистка старых ревизий. Последняя ревизия и раскатанные на спикерах не
удаляются.
</CardDescription>
</CardHeader>
<CardContent class="space-y-4">
@@ -123,16 +224,54 @@
/>
</FormField>
{#if parsedRetentionMinutes !== null}
<div class="space-y-1 rounded-md border bg-muted/30 p-4 text-sm">
<p class="font-medium">Оценка очистки по введённому retention</p>
{#if estimateLoading}
<p class="text-muted-foreground">Расчёт…</p>
{:else if estimate}
<p>
Будет удалено ревизий: <strong>{estimate.revision_count}</strong>
</p>
<p>
Освободится ориентировочно: <strong>~{formatBytes(estimate.bytes_estimate)}</strong>
</p>
{#if estimate.prefix_row_count > 0}
<p class="text-xs text-muted-foreground">
Строк префиксов в снимках: {estimate.prefix_row_count}
{#if estimate.orphan_snapshot_count > 0}
· снимков: {estimate.orphan_snapshot_count}
{/if}
</p>
{/if}
{:else}
<p class="text-muted-foreground">Оценка недоступна</p>
{/if}
<p class="pt-1 text-xs text-muted-foreground">
Учитываются те же правила, что при автоочистке: последняя ревизия и раскатанные на
спикерах не удаляются.
</p>
</div>
{/if}
{#if hasValidationErrors}
<p class="text-sm text-destructive">
Есть ошибки в полях. Исправьте их, чтобы сохранить изменения.
</p>
{/if}
<Button onclick={save} disabled={!canSave}>
<Save />
{saving ? 'Сохранение…' : 'Применить параметры ревизий'}
</Button>
<div class="flex flex-wrap gap-2">
<Button onclick={save} disabled={!canSave}>
<Save />
{saving ? 'Сохранение…' : 'Применить параметры ревизий'}
</Button>
{#if isOperator}
<Button variant="destructive" disabled={!canPruneNow} onclick={requestPruneNow}>
<Trash2 />
{pruning ? 'Очистка…' : 'Очистить сейчас'}
</Button>
{/if}
</div>
{/if}
</CardContent>
</Card>
@@ -0,0 +1,30 @@
import { apiJSON, apiMutate } from '$lib/api/client.js';
export type RevisionPruneEstimate = {
retention_minutes: number;
cutoff_at: string;
revision_count: number;
prefix_row_count: number;
orphan_snapshot_count: number;
bytes_estimate: number;
};
export type RevisionPruneResult = {
deleted_revisions: number;
deleted_prefix_snapshots: number;
deleted_prefix_rows: number;
bytes_estimate: number;
};
export async function fetchRevisionPruneEstimate(
retentionMinutes: number
): Promise<RevisionPruneEstimate> {
const q = new URLSearchParams({ retention_minutes: String(retentionMinutes) });
return apiJSON<RevisionPruneEstimate>(`/v1/revisions/prune-estimate?${q}`);
}
export async function pruneRevisionsNow(retentionMinutes: number): Promise<RevisionPruneResult> {
return apiMutate<RevisionPruneResult>('/v1/revisions/prune', 'POST', {
retention_minutes: retentionMinutes
});
}