internal/app/maintenance/retention.go
Rama main · réplica de lectura del repositorio original.
package maintenance
import (
"context"
"time"
"go.uber.org/zap"
)
// DispatchOutboxRetentionStore cleans up outbox dead tasks that failed permanently (retries abandoned).
type DispatchOutboxRetentionStore interface {
DeleteFailed(ctx context.Context, olderThan time.Duration, limit int) (int, error)
}
// TempAuthKeyRetentionStore reclaims expired PFS temp auth keys (including unbound keys).
type TempAuthKeyRetentionStore interface {
DeleteExpired(ctx context.Context, expiredBefore int64, limit int) (int, error)
}
// AuthKeySessionLayerRetentionStore reclaims expired short-lived Layer
// watermarks. Selector freshness, not retention timing, is the correctness
// gate; this worker only bounds durable storage.
type AuthKeySessionLayerRetentionStore interface {
DeleteExpiredSessionLayers(ctx context.Context, limit int) (int, error)
}
// OrphanAuthKeyRetentionStore reclaims old handshake keys that never formed an
// authorization/temp binding. protected is a snapshot of the raw auth_key_ids actually in use
// by the current connection registry.
type OrphanAuthKeyRetentionStore interface {
DeleteOrphaned(ctx context.Context, olderThan time.Duration, limit int, protected [][8]byte) (int, error)
}
type ActiveRawAuthKeyProvider interface {
ActiveRawAuthKeyIDs() [][8]byte
}
// ActiveAuthKeyHeartbeatStore persists the liveness of raw auth keys still in use by this
// instance. With multiple instances, orphan GC cannot rely only on the current process's
// active snapshot; other instances' heartbeats advance last_used_at in the database so they
// are not misjudged as orphans.
type ActiveAuthKeyHeartbeatStore interface {
TouchActiveRawAuthKeys(ctx context.Context, ids [][8]byte) error
}
// BotAPIUpdateRetentionStore reclaims dead rows of the Bot API getUpdates delivery queue
// (performance audit H1): rows that are confirmed and past the grace period + rows past
// retention by message date (official Bot API updates are kept at most 24h).
type BotAPIUpdateRetentionStore interface {
DeleteDeliveredOrExpired(ctx context.Context, confirmedGrace, maxAge time.Duration, limit int) (int, error)
}
// UserUpdateEventRetentionStore reclaims only the prefix of account events explicitly
// confirmed by all current authorizations. It is not an ordinary TTL: when any authorization
// lacks state the confirmation watermark is 0, and events that device may still need must not
// be deleted.
type UserUpdateEventRetentionStore interface {
DeleteConfirmedPrefix(ctx context.Context, olderThan time.Duration, limit int) (int, error)
}
// ChannelUpdateEventRetentionStore reclaims the consecutive prefix of channel durable
// updates past retention. The concrete store must delete events and advance the retained
// floor in the same transaction; clients below the floor recover via updates.getChannelDifference
// using the channelDifferenceTooLong snapshot.
type ChannelUpdateEventRetentionStore interface {
DeleteExpiredChannelUpdateEvents(ctx context.Context, olderThan time.Duration, limit int) (int, error)
}
// LoginCodeDeliveryRetentionStore reclaims only compact idempotency receipts
// after their associated opaque code lifetime. It must not delete the message,
// durable update event, or outbox facts created by the delivery transaction.
type LoginCodeDeliveryRetentionStore interface {
DeleteExpiredLoginCodeDeliveries(ctx context.Context, expiredBefore time.Time, limit int) (int, error)
}
type ClientTelemetryRetentionStore interface {
DeleteExpiredClientTelemetry(ctx context.Context, olderThan time.Time, limit int) (int, error)
}
type AuthDeliveryReportRetentionStore interface {
DeleteExpiredAuthDeliveryReports(ctx context.Context, olderThan time.Time, limit int) (int, error)
}
type ModerationRetentionStore interface {
DeleteExpiredSponsoredMessageImpressions(ctx context.Context, olderThan time.Time, limit int) (int, error)
DeleteExpiredModerationAppealLinks(ctx context.Context, olderThan time.Time, limit int) (int, error)
}
// botAPIConfirmedGrace is the deletion grace for confirmed Bot API update rows: rows below
// the confirmation watermark are no longer read by getUpdates (fromID is always > confirmed);
// the grace only defends against offset rollback debugging, and the reclamation goal is to
// clear the backlog.
const botAPIConfirmedGrace = 15 * time.Minute
// tempAuthKeyExpiryGrace is only a physical database reclamation grace. The MTProto edge
// stops inbound RPC, active pushes, and retransmissions and disconnects at expires_at;
// ResolveAuthKey does not tolerate expired keys. Deleting a day late absorbs client
// rotation/diagnostics windows and does not extend the protocol validity.
const tempAuthKeyExpiryGrace = 24 * time.Hour
const (
// A terminal failed outbox only provides short-term diagnostic isolation; it is not a
// durable update log. Deleting the task immediately releases the same account's next pts
// through the head trigger, while user_update_events are retained and any missed online
// pushes are compensated by the normal difference path.
defaultOutboxPoisonRetention = time.Minute
defaultOutboxPoisonInterval = 15 * time.Second
)
// RetentionWorker periodically reclaims dead data from storage.
//
// Note: TDesktop does not support account-level updates.differenceTooLong (api_updates.cpp
// only logs the response and does not clear requesting, permanently locking up the update
// engine), so user_update_events must never be hard-trimmed by ordinary TTL. This worker only
// allows the store to delete the consecutive safe prefix that "all current authorized devices
// have explicitly confirmed"; any lagging device or one lacking state pushes the floor back
// to 0. When a client happens to carry back an old pts from before confirmation, the updates
// service advances via the normal differenceSlice checkpoint and does not send
// differenceTooLong.
type RetentionWorker struct {
outbox DispatchOutboxRetentionStore
tempKeys TempAuthKeyRetentionStore // may be nil (does not reclaim temp key bindings)
authKeySessionLayers AuthKeySessionLayerRetentionStore
botAPIUpdates BotAPIUpdateRetentionStore // may be nil (does not reclaim the Bot API queue)
userUpdates UserUpdateEventRetentionStore
channelUpdates ChannelUpdateEventRetentionStore
loginCodeDeliveries LoginCodeDeliveryRetentionStore
clientTelemetry ClientTelemetryRetentionStore
authDeliveryReports AuthDeliveryReportRetentionStore
moderation ModerationRetentionStore
orphanAuthKeys OrphanAuthKeyRetentionStore
activeAuthKeys ActiveRawAuthKeyProvider
activeAuthKeyHeartbeat ActiveAuthKeyHeartbeatStore
logger *zap.Logger
retention time.Duration
botAPIRetention time.Duration
orphanRetention time.Duration
clientTelemetryRetention time.Duration
authDeliveryReportRetention time.Duration
outboxPoisonRetention time.Duration
outboxPoisonInterval time.Duration
interval time.Duration
batch int
}
func NewRetentionWorker(outbox DispatchOutboxRetentionStore, tempKeys TempAuthKeyRetentionStore, logger *zap.Logger, retention, interval time.Duration, batch int) *RetentionWorker {
if logger == nil {
logger = zap.NewNop()
}
if retention <= 0 {
retention = 168 * time.Hour
}
if interval <= 0 {
interval = time.Hour
}
if batch <= 0 {
batch = 10000
}
return &RetentionWorker{
outbox: outbox,
tempKeys: tempKeys,
logger: logger,
retention: retention,
outboxPoisonRetention: defaultOutboxPoisonRetention,
outboxPoisonInterval: defaultOutboxPoisonInterval,
interval: interval,
batch: batch,
}
}
// WithDispatchOutboxPoisonPolicy configures an independent short isolation and cleanup
// period for a terminal failed head. This period cannot reuse the durable update's weekly
// retention, otherwise one deterministic construction error would freeze that user's entire
// online pts lane. Values <=0 fall back to the safe defaults of 1m/15s respectively.
func (w *RetentionWorker) WithDispatchOutboxPoisonPolicy(retention, interval time.Duration) *RetentionWorker {
if retention <= 0 {
retention = defaultOutboxPoisonRetention
}
if interval <= 0 {
interval = defaultOutboxPoisonInterval
}
w.outboxPoisonRetention = retention
w.outboxPoisonInterval = interval
return w
}
// WithBotAPIUpdateRetention enables bot_api_updates queue reclamation; when retention <=0 the official-semantics default of 24h is used.
func (w *RetentionWorker) WithBotAPIUpdateRetention(store BotAPIUpdateRetentionStore, retention time.Duration) *RetentionWorker {
if retention <= 0 {
retention = 24 * time.Hour
}
w.botAPIUpdates = store
w.botAPIRetention = retention
return w
}
// WithUserUpdateRetention enables common-confirmation safe-prefix reclamation for account
// updates. TDesktop does not support account differenceTooLong, so the concrete store must
// guarantee that the unconfirmed prefix is never deleted.
func (w *RetentionWorker) WithUserUpdateRetention(store UserUpdateEventRetentionStore) *RetentionWorker {
w.userUpdates = store
return w
}
// WithChannelUpdateRetention enables bounded TTL reclamation of channel durable updates;
// it reuses the worker's retention/interval/batch, and the store's retained floor guarantees
// that old pts do not read a silent hole.
func (w *RetentionWorker) WithChannelUpdateRetention(store ChannelUpdateEventRetentionStore) *RetentionWorker {
w.channelUpdates = store
return w
}
// WithLoginCodeDeliveryRetention enables bounded seek cleanup for compact
// phone_code_hash receipts. Each row carries its own expiry derived from the
// code TTL, so this cleanup intentionally does not reuse update-log retention.
func (w *RetentionWorker) WithLoginCodeDeliveryRetention(store LoginCodeDeliveryRetentionStore) *RetentionWorker {
w.loginCodeDeliveries = store
return w
}
// WithAuthKeySessionLayerRetention enables bounded seek cleanup for expired
// per-session Layer evidence.
func (w *RetentionWorker) WithAuthKeySessionLayerRetention(store AuthKeySessionLayerRetentionStore) *RetentionWorker {
w.authKeySessionLayers = store
return w
}
func (w *RetentionWorker) WithClientTelemetryRetention(store ClientTelemetryRetentionStore, retention time.Duration) *RetentionWorker {
if retention <= 0 {
retention = 30 * 24 * time.Hour
}
w.clientTelemetry = store
w.clientTelemetryRetention = retention
return w
}
func (w *RetentionWorker) WithAuthDeliveryReportRetention(store AuthDeliveryReportRetentionStore, retention time.Duration) *RetentionWorker {
if retention <= 0 {
retention = 30 * 24 * time.Hour
}
w.authDeliveryReports = store
w.authDeliveryReportRetention = retention
return w
}
func (w *RetentionWorker) WithModerationRetention(store ModerationRetentionStore) *RetentionWorker {
w.moderation = store
return w
}
// WithOrphanAuthKeyRetention enables bounded reclamation of unauthorized handshake keys.
// active must provide raw keys, not temp→perm business keys; otherwise unauthenticated or
// PFS connections would be misjudged as orphans.
func (w *RetentionWorker) WithOrphanAuthKeyRetention(store OrphanAuthKeyRetentionStore, active ActiveRawAuthKeyProvider, retention time.Duration) *RetentionWorker {
w.orphanAuthKeys = store
w.activeAuthKeys = active
w.activeAuthKeyHeartbeat, _ = store.(ActiveAuthKeyHeartbeatStore)
w.orphanRetention = retention
return w
}
func (w *RetentionWorker) Run(ctx context.Context) {
w.runOnce(ctx)
retentionTicker := time.NewTicker(w.interval)
defer retentionTicker.Stop()
poisonTicker := time.NewTicker(w.outboxPoisonInterval)
defer poisonTicker.Stop()
var (
heartbeatTicker *time.Ticker
heartbeatC <-chan time.Time
)
if interval := w.orphanHeartbeatInterval(); interval > 0 {
heartbeatTicker = time.NewTicker(interval)
heartbeatC = heartbeatTicker.C
defer heartbeatTicker.Stop()
}
for {
select {
case <-ctx.Done():
return
case <-retentionTicker.C:
w.runRetentionOnce(ctx)
case <-poisonTicker.C:
w.runOutboxPoisonOnce(ctx)
case <-heartbeatC:
w.heartbeatActiveAuthKeys(ctx)
}
}
}
func (w *RetentionWorker) runOnce(ctx context.Context) {
w.runOutboxPoisonOnce(ctx)
w.runRetentionOnce(ctx)
}
func (w *RetentionWorker) runOutboxPoisonOnce(ctx context.Context) {
if w.outbox == nil {
return
}
outboxDeleted, err := w.outbox.DeleteFailed(ctx, w.outboxPoisonRetention, w.batch)
if err != nil {
w.logger.Error("failed to clean up terminal failed dispatch_outbox",
zap.String("signal", "dispatch_outbox_poison_cleanup_failed"),
zap.Duration("quarantine", w.outboxPoisonRetention),
zap.Error(err),
)
} else if outboxDeleted > 0 {
// The Error-level structured signal is deliberately kept: a terminal failure means a
// deterministic encoding, missing event, or other non-retryable fault. Deleting the task
// only unfreezes the online lane and does not delete the durable event.
w.logger.Error("terminal failed dispatch_outbox isolation ended and user lane released",
zap.String("signal", "dispatch_outbox_poison_released"),
zap.Int("deleted", outboxDeleted),
zap.Duration("quarantine", w.outboxPoisonRetention),
)
}
}
func (w *RetentionWorker) runRetentionOnce(ctx context.Context) {
if w.authKeySessionLayers != nil {
deleted, err := w.authKeySessionLayers.DeleteExpiredSessionLayers(ctx, w.batch)
if err != nil {
w.logger.Warn("failed to reclaim expired auth-key session Layer evidence", zap.Error(err))
} else if deleted > 0 {
w.logger.Info("reclaimed expired auth-key session Layer evidence", zap.Int("deleted", deleted))
}
}
if w.loginCodeDeliveries != nil {
deleted, err := w.loginCodeDeliveries.DeleteExpiredLoginCodeDeliveries(ctx, time.Now(), w.batch)
if err != nil {
w.logger.Warn("failed to reclaim expired login-code delivery receipts", zap.Error(err))
} else if deleted > 0 {
w.logger.Info("reclaimed expired login-code delivery receipts", zap.Int("deleted", deleted))
}
}
if w.clientTelemetry != nil {
deleted, err := w.clientTelemetry.DeleteExpiredClientTelemetry(
ctx, time.Now().Add(-w.clientTelemetryRetention), w.batch,
)
if err != nil {
w.logger.Warn("failed to reclaim expired client telemetry", zap.Error(err))
} else if deleted > 0 {
w.logger.Info("reclaimed expired client telemetry", zap.Int("deleted", deleted))
}
}
if w.authDeliveryReports != nil {
deleted, err := w.authDeliveryReports.DeleteExpiredAuthDeliveryReports(
ctx, time.Now().Add(-w.authDeliveryReportRetention), w.batch,
)
if err != nil {
w.logger.Warn("failed to reclaim expired verification-code delivery diagnostics", zap.Error(err))
} else if deleted > 0 {
w.logger.Info("reclaimed expired verification-code delivery diagnostics", zap.Int("deleted", deleted))
}
}
if w.moderation != nil {
now := time.Now()
impressions, err := w.moderation.DeleteExpiredSponsoredMessageImpressions(
ctx, now, w.batch,
)
if err != nil {
w.logger.Warn("failed to reclaim expired sponsored impressions", zap.Error(err))
} else if impressions > 0 {
w.logger.Info("reclaimed expired sponsored impressions", zap.Int("deleted", impressions))
}
links, err := w.moderation.DeleteExpiredModerationAppealLinks(
ctx, now, w.batch,
)
if err != nil {
w.logger.Warn("failed to reclaim expired moderation appeal links", zap.Error(err))
} else if links > 0 {
w.logger.Info("reclaimed expired moderation appeal links", zap.Int("deleted", links))
}
}
if w.tempKeys != nil {
expiredBefore := time.Now().Add(-tempAuthKeyExpiryGrace).Unix()
tempDeleted, err := w.tempKeys.DeleteExpired(ctx, expiredBefore, w.batch)
if err != nil {
w.logger.Warn("failed to reclaim expired temp auth key bindings", zap.Error(err))
} else if tempDeleted > 0 {
w.logger.Info("reclaimed expired temp auth key bindings", zap.Int("deleted", tempDeleted))
}
}
if w.orphanAuthKeys != nil && w.orphanRetention > 0 {
var protected [][8]byte
if w.activeAuthKeys != nil {
protected = w.activeAuthKeys.ActiveRawAuthKeyIDs()
}
if !w.touchActiveAuthKeys(ctx, protected) {
// Fail safe: if this instance cannot publish its own active set, deleting against a
// stale database heartbeat could evict keys used by another instance too. Keep all
// candidates for this pass and retry after the next heartbeat.
} else {
orphanDeleted, err := w.orphanAuthKeys.DeleteOrphaned(ctx, w.orphanRetention, w.batch, protected)
if err != nil {
w.logger.Warn("failed to reclaim unauthorized orphan auth keys", zap.Error(err))
} else if orphanDeleted > 0 {
w.logger.Info("reclaimed unauthorized orphan auth keys", zap.Int("deleted", orphanDeleted))
}
}
}
if w.botAPIUpdates != nil {
botAPIDeleted, err := w.botAPIUpdates.DeleteDeliveredOrExpired(ctx, botAPIConfirmedGrace, w.botAPIRetention, w.batch)
if err != nil {
w.logger.Warn("failed to reclaim bot_api_updates queue", zap.Error(err))
} else if botAPIDeleted > 0 {
w.logger.Info("reclaimed bot_api_updates queue", zap.Int("deleted", botAPIDeleted))
}
}
if w.userUpdates != nil {
userDeleted, err := w.userUpdates.DeleteConfirmedPrefix(ctx, w.retention, w.batch)
if err != nil {
w.logger.Warn("failed to reclaim commonly confirmed user_update_events prefix", zap.Error(err))
} else if userDeleted > 0 {
w.logger.Info("reclaimed commonly confirmed user_update_events prefix", zap.Int("deleted", userDeleted))
}
}
if w.channelUpdates != nil {
channelDeleted, err := w.channelUpdates.DeleteExpiredChannelUpdateEvents(ctx, w.retention, w.batch)
if err != nil {
// The store isolates bad gaps per channel and continues this round; deleted may be
// non-zero and must be logged at the same time — neither disguising a global pass as a
// total failure nor swallowing invariant errors.
w.logger.Warn("expired channel_update_events reclamation has isolated channels",
zap.Int("deleted", channelDeleted),
zap.Error(err),
)
} else if channelDeleted > 0 {
w.logger.Info("reclaimed expired channel_update_events consecutive prefix", zap.Int("deleted", channelDeleted))
}
}
}
func (w *RetentionWorker) orphanHeartbeatInterval() time.Duration {
if w.activeAuthKeyHeartbeat == nil || w.activeAuthKeys == nil || w.orphanRetention <= 0 {
return 0
}
interval := w.orphanRetention / 3
if interval <= 0 {
interval = time.Nanosecond
}
if w.interval > 0 && w.interval < interval {
interval = w.interval
}
return interval
}
func (w *RetentionWorker) heartbeatActiveAuthKeys(ctx context.Context) {
if w.activeAuthKeys == nil {
return
}
w.touchActiveAuthKeys(ctx, w.activeAuthKeys.ActiveRawAuthKeyIDs())
}
// touchActiveAuthKeys returns false only when a configured durable heartbeat failed. A store that
// predates the optional heartbeat interface keeps single-instance behavior.
func (w *RetentionWorker) touchActiveAuthKeys(ctx context.Context, protected [][8]byte) bool {
if w.activeAuthKeyHeartbeat == nil {
return true
}
if err := w.activeAuthKeyHeartbeat.TouchActiveRawAuthKeys(ctx, protected); err != nil {
w.logger.Error("failed to refresh active raw auth key heartbeat, skipping orphan GC this round",
zap.String("signal", "auth_key_heartbeat_failed"),
zap.Int("active_keys", len(protected)),
zap.Error(err),
)
return false
}
return true
}