feat: add debug logging and optimize database queries in Postgres repository
CI / changes (push) Successful in 6s
CI / openapi (push) Has been skipped
CI / go (push) Successful in 40s
CI / docker-web (deploy/docker/evobgp-web/Dockerfile, , evobgp-web) (push) Has been skipped
CI / docker-web (deploy/docker/evobgp-web/Dockerfile, evobgp-all, evobgp-web-all) (push) Has been skipped
CI / docker-bird (push) Has been skipped
CI / bird2 (push) Successful in 17s
CI / docker-go-prime (push) Successful in 25s
CI / docker-go (deploy/docker/evobgp-agent/Dockerfile, , evobgp-agent) (push) Successful in 1m3s
CI / docker-go (evobgp-all, 1, deploy/docker/gobinary/Dockerfile, , evobgp-all) (push) Successful in 3m4s
CI / docker-go (evobgp-api, 1, deploy/docker/gobinary/Dockerfile, , evobgp-api) (push) Successful in 1m22s
CI / docker-go (evobgp-deploy, 0, deploy/docker/gobinary/Dockerfile, , evobgp-deploy) (push) Successful in 1m22s
CI / docker-go (evobgp-ingest, 0, deploy/docker/gobinary/Dockerfile, , evobgp-ingest) (push) Successful in 1m23s
CI / docker-go (evobgp-node, 0, deploy/docker/gobinary/Dockerfile, , evobgp-node) (push) Successful in 1m10s
CI / docker-go (evobgp-render, 0, deploy/docker/gobinary/Dockerfile, , evobgp-render) (push) Successful in 1m22s
CI / docker-go (evobgp-scheduler, 0, deploy/docker/gobinary/Dockerfile, , evobgp-scheduler) (push) Successful in 1m25s

Introduced a new debug logging function to capture detailed information during specific operations, controlled by an environment variable. Optimized the MaterializedPrefixStats and ListRevisions methods to reduce memory usage by leveraging SQL aggregation and limiting result sets. Updated the ListRevisionPrefixes method to include pagination support, enhancing performance and efficiency in data retrieval.
This commit is contained in:
Denozordec
2026-04-08 12:51:15 +07:00
parent f6b94a44d0
commit 6d0e214655
+134 -82
View File
@@ -6,7 +6,8 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"sort" "os"
"runtime"
"strconv" "strconv"
"strings" "strings"
"time" "time"
@@ -18,6 +19,36 @@ import (
"github.com/jackc/pgx/v5/pgxpool" "github.com/jackc/pgx/v5/pgxpool"
) )
// #region agent log
func agentDebugNDJSON3214(hypothesisID, location, message string, data map[string]any) {
if os.Getenv("EVOBGP_DEBUG_LOG") != "1" {
return
}
f, err := os.OpenFile("debug-3214dc.log", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return
}
defer f.Close()
var ms runtime.MemStats
runtime.ReadMemStats(&ms)
payload := map[string]any{
"sessionId": "3214dc",
"hypothesisId": hypothesisID,
"location": location,
"message": message,
"data": data,
"timestamp": time.Now().UnixMilli(),
"allocBytes": ms.Alloc,
}
b, err := json.Marshal(payload)
if err != nil {
return
}
_, _ = f.Write(append(b, '\n'))
}
// #endregion
// Postgres implements store.Backend using pgxpool. // Postgres implements store.Backend using pgxpool.
type Postgres struct { type Postgres struct {
pool *pgxpool.Pool pool *pgxpool.Pool
@@ -42,22 +73,15 @@ func (p *Postgres) DemoIDs() (tenant, moduleCDN, moduleIP, revision, speaker str
func (p *Postgres) MaterializedPrefixStats() (max int, sum int) { func (p *Postgres) MaterializedPrefixStats() (max int, sum int) {
ctx := context.Background() ctx := context.Background()
rows, err := p.pool.Query(ctx, ` // Агрегация в БД — не тащим все строки config_revision в память.
SELECT COALESCE((meta_json->>'materialized_prefix_count')::int, 0) AS n err := p.pool.QueryRow(ctx, `
FROM config_revision`) SELECT
COALESCE(MAX((meta_json->>'materialized_prefix_count')::int), 0),
COALESCE(SUM((meta_json->>'materialized_prefix_count')::int), 0)
FROM config_revision`).Scan(&max, &sum)
if err != nil { if err != nil {
return 0, 0 return 0, 0
} }
defer rows.Close()
for rows.Next() {
var n int
if rows.Scan(&n) == nil {
sum += n
if n > max {
max = n
}
}
}
return max, sum return max, sum
} }
@@ -554,14 +578,29 @@ func (p *Postgres) ListRevisions(tenantID, moduleID string, cursor string, limit
if limit <= 0 { if limit <= 0 {
limit = 50 limit = 50
} }
off := 0
if cursor != "" {
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
off = n
}
}
ctx := context.Background() ctx := context.Background()
q := `SELECT id::text, module_id::text, content_hash, parent_revision_id::text, meta_json, created_at FROM config_revision WHERE tenant_id=$1` // List endpoint only needs materialized_prefix_count from meta_json — not full preview_fragments blobs.
// LIMIT/OFFSET in SQL avoids loading every revision for the tenant into memory (was O(N) per request).
q := `SELECT id::text, module_id::text, content_hash, parent_revision_id::text,
COALESCE((meta_json->>'materialized_prefix_count')::int, 0), created_at
FROM config_revision WHERE tenant_id=$1`
args := []any{tenantID} args := []any{tenantID}
n := 2
if moduleID != "" { if moduleID != "" {
q += ` AND module_id = $2` q += fmt.Sprintf(` AND module_id=$%d`, n)
args = append(args, moduleID) args = append(args, moduleID)
n++
} }
q += ` ORDER BY created_at DESC` q += ` ORDER BY created_at DESC`
// Fetch limit+1 rows to compute has_more without COUNT(*).
q += fmt.Sprintf(` LIMIT $%d OFFSET $%d`, n, n+1)
args = append(args, limit+1, off)
rows, err := p.pool.Query(ctx, q, args...) rows, err := p.pool.Query(ctx, q, args...)
if err != nil { if err != nil {
return nil, "", false return nil, "", false
@@ -572,58 +611,53 @@ func (p *Postgres) ListRevisions(tenantID, moduleID string, cursor string, limit
var r store.Revision var r store.Revision
r.TenantID = tenantID r.TenantID = tenantID
var mod, parent *string var mod, parent *string
var meta []byte var mpc int
if err := rows.Scan(&r.ID, &mod, &r.ContentHash, &parent, &meta, &r.CreatedAt); err != nil { if err := rows.Scan(&r.ID, &mod, &r.ContentHash, &parent, &mpc, &r.CreatedAt); err != nil {
continue continue
} }
if mod != nil { if mod != nil {
r.ModuleID = *mod r.ModuleID = *mod
} }
r.ParentRevisionID = strOrNil(parent) r.ParentRevisionID = strOrNil(parent)
var mj struct { r.MaterializedPrefixCount = mpc
PreviewFragments map[string]string `json:"preview_fragments"` r.PreviewFragments = map[string]string{}
MaterializedPrefixCount int `json:"materialized_prefix_count"`
}
_ = json.Unmarshal(meta, &mj)
if mj.PreviewFragments == nil {
mj.PreviewFragments = map[string]string{}
}
r.PreviewFragments = mj.PreviewFragments
r.MaterializedPrefixCount = mj.MaterializedPrefixCount
all = append(all, &r) all = append(all, &r)
} }
agentDebugNDJSON3214("A", "repository/postgres.go:ListRevisions", "list_revisions_fetched", map[string]any{
"rows": len(all), "limit": limit, "offset": off,
})
hasMore := len(all) > limit
if hasMore {
all = all[:limit]
}
next := ""
if hasMore {
next = fmt.Sprintf("%d", off+limit)
}
if len(all) == 0 {
return nil, "", false
}
return all, next, hasMore
}
func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor string, limit int) ([]store.PrefixRow, string, bool) {
if limit <= 0 {
limit = 50
}
off := 0 off := 0
if cursor != "" { if cursor != "" {
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 { if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
off = n off = n
} }
} }
end := off + limit
next := ""
hasMore := false
if end > len(all) {
end = len(all)
} else {
hasMore = true
next = fmt.Sprintf("%d", end)
}
if off >= len(all) {
return nil, "", false
}
return all[off:end], next, hasMore
}
func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor string, limit int) ([]store.PrefixRow, string, bool) {
if limit <= 0 {
limit = 50
}
ctx := context.Background() ctx := context.Background()
if _, err := p.GetRevision(tenantID, revisionID); err != nil { if _, err := p.GetRevision(tenantID, revisionID); err != nil {
return nil, "", false return nil, "", false
} }
rows, err := p.pool.Query(ctx, ` rows, err := p.pool.Query(ctx, `
SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix
WHERE revision_id=$1 ORDER BY id`, revisionID) WHERE revision_id=$1 ORDER BY id
LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
if err != nil { if err != nil {
return nil, "", false return nil, "", false
} }
@@ -638,25 +672,21 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
pr.CommunityID = comm pr.CommunityID = comm
all = append(all, pr) all = append(all, pr)
} }
off := 0 agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{
if cursor != "" { "rows": len(all), "limit": limit, "offset": off,
if n, err := strconv.Atoi(cursor); err == nil && n >= 0 { })
off = n more := len(all) > limit
} if more {
all = all[:limit]
} }
end := off + limit
next := "" next := ""
more := false if more {
if end > len(all) { next = fmt.Sprintf("%d", off+limit)
end = len(all)
} else {
more = true
next = fmt.Sprintf("%d", end)
} }
if off >= len(all) { if len(all) == 0 {
return nil, "", false return nil, "", false
} }
return all[off:end], next, more return all, next, more
} }
func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (string, error) { func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (string, error) {
@@ -697,32 +727,54 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
if _, err := p.GetRevision(tenantID, bID); err != nil { if _, err := p.GetRevision(tenantID, bID); err != nil {
return nil, err return nil, err
} }
pa, _, _ := p.ListRevisionPrefixes(tenantID, aID, "", 100000) ctx := context.Background()
pb, _, _ := p.ListRevisionPrefixes(tenantID, bID, "", 100000) var unchanged int
setA := make(map[string]struct{}) err := p.pool.QueryRow(ctx, `
setB := make(map[string]struct{}) SELECT COUNT(*)::int FROM (
for _, x := range pa { SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
setA[x.Prefix] = struct{}{} INTERSECT
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
) t`, aID, bID).Scan(&unchanged)
if err != nil {
return nil, err
} }
for _, x := range pb { // added: в B, нет в A; removed: в A, нет в B — без загрузки полных снапшотов в память.
setB[x.Prefix] = struct{}{} rowsAdded, err := p.pool.Query(ctx, `
SELECT prefix::text FROM (
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
EXCEPT
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
) s ORDER BY 1`, bID, aID)
if err != nil {
return nil, err
} }
var added, removed []string defer rowsAdded.Close()
unchanged := 0 var added []string
for pfx := range setB { for rowsAdded.Next() {
if _, ok := setA[pfx]; !ok { var s string
added = append(added, pfx) if err := rowsAdded.Scan(&s); err != nil {
} else { continue
unchanged++
} }
added = append(added, s)
} }
for pfx := range setA { rowsRem, err := p.pool.Query(ctx, `
if _, ok := setB[pfx]; !ok { SELECT prefix::text FROM (
removed = append(removed, pfx) SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
EXCEPT
SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
) s ORDER BY 1`, aID, bID)
if err != nil {
return nil, err
}
defer rowsRem.Close()
var removed []string
for rowsRem.Next() {
var s string
if err := rowsRem.Scan(&s); err != nil {
continue
} }
removed = append(removed, s)
} }
sort.Strings(added)
sort.Strings(removed)
return map[string]any{ return map[string]any{
"revision_a": aID, "revision_a": aID,
"revision_b": bID, "revision_b": bID,