DenozordecandCursor
f8e0520817
Очередь job получила lease и fencing, телеметрия и прогресс deploy сохраняются, ревизия пишет manifest входов. Обновление префиксов не переписывает неизменённый снимок, большие наборы идут через COPY, а устаревшая ревизия не попадает в deploy. Бандл ноды кэшируется, prune и постановка job больше не держат горячий путь. Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
2.5 KiB
Go
109 lines
2.5 KiB
Go
package pipeline
|
|
|
|
import (
|
|
"context"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"evobgp/internal/store"
|
|
)
|
|
|
|
const (
|
|
revisionPruneMinInterval = 5 * time.Minute
|
|
revisionPruneTimeout = 2 * time.Minute
|
|
)
|
|
|
|
// revisionPruneGate runs at most one prune per tenant per interval and drops
|
|
// overlapping attempts so renders do not stack prune transactions.
|
|
type revisionPruneGate struct {
|
|
mu sync.Mutex
|
|
last map[string]time.Time
|
|
inflight map[string]chan struct{}
|
|
now func() time.Time
|
|
}
|
|
|
|
var revisionPrunes revisionPruneGate
|
|
|
|
func (g *revisionPruneGate) start(tenantID string) bool {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
now := time.Now()
|
|
if g.now != nil {
|
|
now = g.now()
|
|
}
|
|
if g.last == nil {
|
|
g.last = make(map[string]time.Time)
|
|
g.inflight = make(map[string]chan struct{})
|
|
}
|
|
if _, running := g.inflight[tenantID]; running {
|
|
return false
|
|
}
|
|
if last, ok := g.last[tenantID]; ok && now.Sub(last) < revisionPruneMinInterval {
|
|
return false
|
|
}
|
|
g.last[tenantID] = now
|
|
g.inflight[tenantID] = make(chan struct{})
|
|
return true
|
|
}
|
|
|
|
func (g *revisionPruneGate) done(tenantID string) {
|
|
g.mu.Lock()
|
|
defer g.mu.Unlock()
|
|
ch, ok := g.inflight[tenantID]
|
|
if !ok {
|
|
return
|
|
}
|
|
close(ch)
|
|
delete(g.inflight, tenantID)
|
|
}
|
|
|
|
// applyRevisionRetention schedules retention for tenantID and returns without
|
|
// waiting. Prune uses its own short context, at most once per tenant per five
|
|
// minutes, and overlapping renders do not start another transaction.
|
|
func applyRevisionRetention(st store.Backend, tenantID string) {
|
|
tenantID = strings.TrimSpace(tenantID)
|
|
if st == nil || tenantID == "" {
|
|
return
|
|
}
|
|
if !revisionPrunes.start(tenantID) {
|
|
return
|
|
}
|
|
go func() {
|
|
defer revisionPrunes.done(tenantID)
|
|
ctx, cancel := context.WithTimeout(context.Background(), revisionPruneTimeout)
|
|
defer cancel()
|
|
pruneTenantRevisions(ctx, st, tenantID)
|
|
}()
|
|
}
|
|
|
|
// pruneTenantRevisions applies the tenant retention cutoff immediately.
|
|
func pruneTenantRevisions(ctx context.Context, st store.Backend, tenantID string) {
|
|
if st == nil || ctx == nil {
|
|
return
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return
|
|
}
|
|
settings, err := st.ListGlobalSettings(tenantID)
|
|
if err != nil {
|
|
return
|
|
}
|
|
_, _ = st.PruneRevisionsBefore(ctx, tenantID, RevisionRetentionCutoff(settings))
|
|
}
|
|
|
|
func waitRevisionPrune(ctx context.Context, tenantID string) error {
|
|
revisionPrunes.mu.Lock()
|
|
ch := revisionPrunes.inflight[strings.TrimSpace(tenantID)]
|
|
revisionPrunes.mu.Unlock()
|
|
if ch == nil {
|
|
return nil
|
|
}
|
|
select {
|
|
case <-ch:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|