perf(pipeline): parallel prefetch and lean revision GET

GetRevisionSummary без preview_fragments; boundedRepoCtx для GetRevision;
parallel CDN prefetch с ctx из ingest.

Co-authored-by: Cursor <[email protected]>

internal/httpapi/routes.go

@@ -579,7 +579,7 @@ func (s *Server) handleGetRevision(w http.ResponseWriter, r *http.Request) {
 	if !s.requireAtLeast(w, a, "viewer") {
 		return
 	}
-	rev, err := s.store.GetRevision(a.TenantID, r.PathValue("revision_id"))
+	rev, err := s.store.GetRevisionSummary(a.TenantID, r.PathValue("revision_id"))
 	if err != nil {
 		writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
 		return

internal/ingest/run.go

@@ -3,11 +3,11 @@ package ingest
 import (
 	"context"
 	"log"
-	"net/http"
 	"time"
 
 	"evobgp/internal/broker"
 	"evobgp/internal/config"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/pipeline"
 	"evobgp/internal/store"
 )
@@ -24,7 +24,7 @@ func Run(ctx context.Context, deps *Deps) {
 	if deps == nil || deps.Store == nil {
 		log.Fatalf("evobgp-ingest: missing store (pass ingest.Deps from BootstrapWorkers or evobgp-all)")
 	}
-	hc := &http.Client{Timeout: 45 * time.Second}
+	hc := httpclient.New(httpclient.DefaultTimeout)
 	t := time.NewTicker(60 * time.Second)
 	defer t.Stop()
 	log.Printf("evobgp-ingest: active (CDN conditional GET / ETag prefetch)")
@@ -34,7 +34,10 @@ func Run(ctx context.Context, deps *Deps) {
 			log.Printf("evobgp-ingest: stopped")
 			return
 		case <-t.C:
-			if err := pipeline.PrefetchCDNSourceETags(context.Background(), deps.Store, hc); err != nil {
+			prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second)
+			err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc)
+			cancel()
+			if err != nil {
 				log.Printf("evobgp-ingest: prefetch: %v", err)
 			}
 		}

internal/pipeline/prefetch.go

@@ -5,87 +5,116 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"evobgp/internal/httpclient"
 	"evobgp/internal/store"
 )
 
+type prefetchTask struct {
+	tenantID string
+	mod      *store.Module
+	src      *store.CDNSource
+}
+
 // PrefetchCDNSourceETags performs conditional GETs for CDN sources; on 200 parses CIDRs into module_prefix_snapshot.
 func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Client) error {
 	if hc == nil {
 		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	tenants, err := st.ListTenantIDs()
 	if err != nil {
 		return err
 	}
-	now := time.Now().UTC()
+	var tasks []prefetchTask
 	for _, tid := range tenants {
 		for _, mod := range st.ListModules(tid) {
 			if mod == nil || !mod.Enabled || mod.Type != "CDN_CIDRS" {
 				continue
 			}
-			omod, err := st.GetModule(tid, mod.ID)
-			if err != nil {
-				continue
-			}
 			sources, err := st.ListCDNSources(tid, mod.ID)
 			if err != nil {
 				continue
 			}
-			var prior []store.PrefixRow
-			if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil {
-				prior = snap.Prefixes
-			}
 			for _, src := range sources {
-				if src == nil || strings.TrimSpace(src.URL) == "" {
-					continue
-				}
-				req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSpace(src.URL), nil)
-				if err != nil {
-					continue
-				}
-				if etag := strings.TrimSpace(src.Etag); etag != "" {
-					req.Header.Set("If-None-Match", etag)
-				}
-				resp, err := hc.Do(req)
-				if err != nil {
-					continue
-				}
-				if resp.StatusCode == http.StatusNotModified {
-					_ = resp.Body.Close()
-					continue
-				}
-				if resp.StatusCode != http.StatusOK {
-					_, _ = io.Copy(io.Discard, resp.Body)
-					_ = resp.Body.Close()
-					continue
-				}
-				body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
-				_ = resp.Body.Close()
-				if err != nil {
-					continue
-				}
-				prefixStrs, err := parseCDNBody(string(body), src)
-				if err != nil {
-					continue
-				}
-				newEtag := strings.TrimSpace(resp.Header.Get("ETag"))
-				patch := &store.CDNSourcePatch{LastRefreshedAt: &now}
-				if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) {
-					e := newEtag
-					patch.Etag = &e
-				}
-				_, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch)
-				rows := cdnRowsFromParsed(omod, src, prefixStrs)
-				_ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows)
-				_ = prior // prior may be stale after merge; refresh for next source in loop
-				if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil {
-					prior = snap.Prefixes
+				if src != nil && strings.TrimSpace(src.URL) != "" {
+					tasks = append(tasks, prefetchTask{tenantID: tid, mod: mod, src: src})
 				}
 			}
 		}
 	}
+	if len(tasks) == 0 {
+		return nil
+	}
+	sem := make(chan struct{}, collectConcurrency())
+	var wg sync.WaitGroup
+	for _, task := range tasks {
+		wg.Add(1)
+		go func(t prefetchTask) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			prefetchOneCDNSource(ctx, st, hc, t)
+		}(task)
+	}
+	wg.Wait()
 	return nil
 }
+
+func prefetchOneCDNSource(ctx context.Context, st store.Backend, hc *http.Client, t prefetchTask) {
+	now := time.Now().UTC()
+	tid, mod, src := t.tenantID, t.mod, t.src
+	u := strings.TrimSpace(src.URL)
+	if _, err := ValidateCDNURL(u); err != nil {
+		return
+	}
+	if err := ResolveCDNURLHost(ctx, u); err != nil {
+		return
+	}
+	omod, err := st.GetModule(tid, mod.ID)
+	if err != nil {
+		return
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return
+	}
+	if etag := strings.TrimSpace(src.Etag); etag != "" {
+		req.Header.Set("If-None-Match", etag)
+	}
+	resp, err := upstreamHTTPDo(ctx, hc, req)
+	if err != nil {
+		return
+	}
+	if resp.StatusCode == http.StatusNotModified {
+		_ = resp.Body.Close()
+		return
+	}
+	if resp.StatusCode != http.StatusOK {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+	_ = resp.Body.Close()
+	if err != nil {
+		return
+	}
+	prefixStrs, err := parseCDNBody(string(body), src)
+	if err != nil {
+		return
+	}
+	newEtag := strings.TrimSpace(resp.Header.Get("ETag"))
+	patch := &store.CDNSourcePatch{LastRefreshedAt: &now}
+	if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) {
+		e := newEtag
+		patch.Etag = &e
+	}
+	_, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch)
+	rows := cdnRowsFromParsed(omod, src, prefixStrs)
+	_ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows)
+}

internal/repository/ctx.go

@@ -0,0 +1,19 @@
+package repository
+
+import (
+	"context"
+	"time"
+)
+
+const defaultRepoTimeout = 60 * time.Second
+
+// boundedRepoCtx returns a context with default repository I/O timeout.
+func boundedRepoCtx(parent context.Context) (context.Context, context.CancelFunc) {
+	if parent == nil {
+		parent = context.Background()
+	}
+	if _, ok := parent.Deadline(); ok {
+		return parent, func() {}
+	}
+	return context.WithTimeout(parent, defaultRepoTimeout)
+}

internal/repository/postgres.go

@@ -659,7 +659,8 @@ func (p *Postgres) DeleteSpeaker(tenantID, id string) error {
 }
 
 func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, error) {
-	ctx := context.Background()
+	ctx, cancel := boundedRepoCtx(context.Background())
+	defer cancel()
 	var r store.Revision
 	var mod *string
 	var parent *string
@@ -691,6 +692,33 @@ func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, er
 	return &r, nil
 }
 
+func (p *Postgres) GetRevisionSummary(tenantID, revisionID string) (*store.Revision, error) {
+	ctx, cancel := boundedRepoCtx(context.Background())
+	defer cancel()
+	var r store.Revision
+	var mod *string
+	var parent *string
+	var prefixCount int
+	err := p.pool.QueryRow(ctx, `
+		SELECT id::text, tenant_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 id=$1 AND tenant_id=$2`, revisionID, tenantID).Scan(
+		&r.ID, &r.TenantID, &mod, &r.ContentHash, &parent, &prefixCount, &r.CreatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, store.ErrNotFound
+		}
+		return nil, err
+	}
+	if mod != nil {
+		r.ModuleID = *mod
+	}
+	r.ParentRevisionID = strOrNil(parent)
+	r.MaterializedPrefixCount = prefixCount
+	r.PreviewFragments = map[string]string{}
+	return &r, nil
+}
+
 func (p *Postgres) ListRevisions(tenantID, moduleID string, cursor string, limit int) ([]*store.Revision, string, bool) {
 	if limit <= 0 {
 		limit = 50

internal/store/backend.go

@@ -76,6 +76,8 @@ type Backend interface {
 	DeleteSpeaker(tenantID, id string) error
 
 	GetRevision(tenantID, revisionID string) (*Revision, error)
+	// GetRevisionSummary returns revision metadata without preview_fragments payloads.
+	GetRevisionSummary(tenantID, revisionID string) (*Revision, error)
 	ListRevisions(tenantID, moduleID string, cursor string, limit int) (items []*Revision, nextCursor string, hasMore bool)
 	ListRevisionPrefixes(tenantID, revisionID string, cursor string, limit int) (prefixes []PrefixRow, next string, more bool)
 	CreateRollbackRevision(tenantID, sourceRevisionID string) (newID string, err error)

internal/store/memory.go

@@ -424,14 +424,19 @@ func (m *Memory) GetModule(tenantID, moduleID string) (*Module, error) {
 func (m *Memory) GetRevision(tenantID, revisionID string) (*Revision, error) {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
-	rev, ok := m.revisions[revisionID]
-	if !ok {
-		return nil, ErrNotFound
+	return m.getRevisionLocked(tenantID, revisionID)
+}
+
+func (m *Memory) GetRevisionSummary(tenantID, revisionID string) (*Revision, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	rev, err := m.getRevisionLocked(tenantID, revisionID)
+	if err != nil {
+		return nil, err
 	}
-	if rev.TenantID != tenantID {
-		return nil, ErrTenantScope
-	}
-	return rev, nil
+	cp := *rev
+	cp.PreviewFragments = nil
+	return &cp, nil
 }
 
 func (m *Memory) GetSpeaker(tenantID, speakerID string) (*Speaker, error) {