adding more functions to media managament system
This commit is contained in:
onysd
parent
95e62c2d77
commit
70c0ba44f0
30 changed files with 1494 additions and 104 deletions
|
|
@ -87,8 +87,11 @@ func (c *stickerSetNegativeCache) delete(refs ...domain.StickerSetRef) {
|
|||
// 每个 chunk 一次 GetFileBlob 的 PG 往返（一个文件按 ≤512KB/1MB 分多次 getFile，热门贴纸/
|
||||
// reaction/头像更被大量用户重复拉）。
|
||||
//
|
||||
// FileBlob 元数据小（约百字节）且内容不可变：location_key 一旦写入即固定指向同一 object_key，
|
||||
// 新建 blob 用随机 id 生成 location_key 不会与已缓存项冲突，故只读填充、无需失效。
|
||||
// FileBlob 元数据小（约百字节），新建 blob 用随机 id 生成 location_key 不会与已缓存项冲突，
|
||||
// 故写入路径只读填充、无需失效。但一个 location_key 的 file_blobs 行可以被存储 retention
|
||||
// 清理主动删除（尤其是 hard 模式：不等 orphaned，仍被引用的媒体到期也会被清理字节），这种情况
|
||||
// 下必须调用 delete 使该 key 失效，否则缓存会继续认为它存在，导致后续下载走到已被删除的字节
|
||||
// 而不是优雅返回 LOCATION_INVALID。
|
||||
type blobMetaCache struct {
|
||||
mu sync.Mutex
|
||||
cap int
|
||||
|
|
@ -141,6 +144,17 @@ func (c *blobMetaCache) put(key string, blob domain.FileBlob) {
|
|||
}
|
||||
}
|
||||
|
||||
// delete 使一个 location_key 的缓存元数据立即失效（如果存在）。见上方类型注释：
|
||||
// 存储 retention 清理主动删掉该 location_key 的 file_blobs 行后必须调用。
|
||||
func (c *blobMetaCache) delete(key string) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if el, ok := c.m[key]; ok {
|
||||
c.ll.Remove(el)
|
||||
delete(c.m, key)
|
||||
}
|
||||
}
|
||||
|
||||
// blobBytesCache 是 object_key → 小 blob 全量字节的 LRU。Sticker / reaction /
|
||||
// 缩略图通常只有几 KB 到几十 KB，缓存全量内容可以避开点击历史时的本地磁盘冷读抖动；
|
||||
// 大媒体仍由 BlobBackend.GetRange 分段读取，避免把大文件放进内存。
|
||||
|
|
@ -220,6 +234,21 @@ func (c *blobBytesCache) put(key string, bytes []byte) {
|
|||
}
|
||||
}
|
||||
|
||||
// delete evicts a cached object_key's full bytes (if present) and reclaims
|
||||
// its budget. Called after a retention sweep physically deletes the object
|
||||
// from its backend, so a later GetFile can't keep serving bytes that no
|
||||
// longer exist on disk/S3.
|
||||
func (c *blobBytesCache) delete(key string) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if el, ok := c.m[key]; ok {
|
||||
entry := el.Value.(*blobBytesEntry)
|
||||
c.ll.Remove(el)
|
||||
delete(c.m, key)
|
||||
c.used -= entry.size
|
||||
}
|
||||
}
|
||||
|
||||
type stickerSetFullCache struct {
|
||||
mu sync.RWMutex
|
||||
byID map[int64]stickerSetFullEntry
|
||||
|
|
|
|||
|
|
@ -24,6 +24,17 @@ type mediaRetentionStore interface {
|
|||
// OrphanDocumentIfUnreferenced is the immediate (no grace period)
|
||||
// counterpart to the age-based sweep above -- see its doc comment.
|
||||
OrphanDocumentIfUnreferenced(ctx context.Context, id int64) (bool, error)
|
||||
|
||||
// -- "hard" retention mode (TELESRV_STORAGE_RETENTION_MODE=hard) --
|
||||
// Candidate selection ignores media_references entirely: a document/
|
||||
// photo still referenced by a live message is exactly as eligible as an
|
||||
// orphaned one once it's old enough. The delete methods physically
|
||||
// remove only the file_blobs row(s)/bytes, never the document/photo
|
||||
// metadata row -- see DeleteFileBlobsForDocument's doc comment.
|
||||
ListDocumentIDsForHardRetentionOlderThan(ctx context.Context, cutoff time.Time, limit int) ([]int64, error)
|
||||
ListPhotoIDsForHardRetentionOlderThan(ctx context.Context, cutoff time.Time, limit int) ([]int64, error)
|
||||
DeleteFileBlobsForDocument(ctx context.Context, id int64) ([]domain.FileBlob, error)
|
||||
DeleteFileBlobsForPhoto(ctx context.Context, id int64) ([]domain.FileBlob, error)
|
||||
}
|
||||
|
||||
// DeleteOrphanedOlderThan implements maintenance.OrphanedMediaRetentionStore:
|
||||
|
|
@ -68,6 +79,63 @@ func (s *Service) DeleteOrphanedOlderThan(ctx context.Context, cutoff time.Time,
|
|||
return deleted, nil
|
||||
}
|
||||
|
||||
// DeleteBlobBytesForMediaOlderThan implements
|
||||
// maintenance.HardMediaRetentionStore ("hard" retention mode): for
|
||||
// documents/photos whose upload/created_at is older than cutoff, physically
|
||||
// deletes their blob bytes (main body + thumbnail/rendition variants) from
|
||||
// the backend and removes their file_blobs rows -- REGARDLESS of whether a
|
||||
// live message/profile-photo/sticker-set still references them. It
|
||||
// deliberately never touches the documents/photos metadata row itself: a
|
||||
// message must still be able to render "here was a photo/document"
|
||||
// (dimensions, mime type, filename) after its bytes are gone, rather than
|
||||
// the message breaking outright. A subsequent upload.getFile for the same
|
||||
// location key finds no file_blobs row and returns LOCATION_INVALID, which
|
||||
// stock clients already render as a "media unavailable" placeholder.
|
||||
func (s *Service) DeleteBlobBytesForMediaOlderThan(ctx context.Context, cutoff time.Time, limit int) (int, error) {
|
||||
store, ok := s.media.(mediaRetentionStore)
|
||||
if !ok || limit <= 0 {
|
||||
return 0, nil
|
||||
}
|
||||
purged := 0
|
||||
docIDs, err := store.ListDocumentIDsForHardRetentionOlderThan(ctx, cutoff, limit)
|
||||
if err != nil {
|
||||
return purged, fmt.Errorf("list documents for hard retention: %w", err)
|
||||
}
|
||||
for _, id := range docIDs {
|
||||
blobs, err := store.DeleteFileBlobsForDocument(ctx, id)
|
||||
if err != nil {
|
||||
s.log.Warn("hard-delete document blob bytes failed", zap.Int64("document_id", id), zap.Error(err))
|
||||
continue
|
||||
}
|
||||
if len(blobs) == 0 {
|
||||
continue
|
||||
}
|
||||
s.deleteOrphanedBlobs(ctx, store, blobs)
|
||||
purged++
|
||||
}
|
||||
remaining := limit - len(docIDs)
|
||||
if remaining <= 0 {
|
||||
return purged, nil
|
||||
}
|
||||
photoIDs, err := store.ListPhotoIDsForHardRetentionOlderThan(ctx, cutoff, remaining)
|
||||
if err != nil {
|
||||
return purged, fmt.Errorf("list photos for hard retention: %w", err)
|
||||
}
|
||||
for _, id := range photoIDs {
|
||||
blobs, err := store.DeleteFileBlobsForPhoto(ctx, id)
|
||||
if err != nil {
|
||||
s.log.Warn("hard-delete photo blob bytes failed", zap.Int64("photo_id", id), zap.Error(err))
|
||||
continue
|
||||
}
|
||||
if len(blobs) == 0 {
|
||||
continue
|
||||
}
|
||||
s.deleteOrphanedBlobs(ctx, store, blobs)
|
||||
purged++
|
||||
}
|
||||
return purged, nil
|
||||
}
|
||||
|
||||
// deleteDocumentNowIfUnreferenced is the immediate counterpart to the
|
||||
// age-based sweep DeleteOrphanedOlderThan runs in the background: orphans
|
||||
// id right now (skipping the grace period) and, only if that succeeds --
|
||||
|
|
@ -106,6 +174,17 @@ func (s *Service) deleteDocumentNowIfUnreferenced(ctx context.Context, id int64)
|
|||
// delete it from.
|
||||
func (s *Service) deleteOrphanedBlobs(ctx context.Context, store mediaRetentionStore, blobs []domain.FileBlob) {
|
||||
for _, b := range blobs {
|
||||
// The file_blobs row for this exact location_key is already gone
|
||||
// (the caller deleted it in the same transaction that produced this
|
||||
// blob list) -- so any cached "found" metadata for it is now wrong
|
||||
// regardless of whether the underlying bytes turn out to still be
|
||||
// shared by another row below. Without this, a hot GetFile path that
|
||||
// had this location_key's metadata cached would keep trying to read
|
||||
// bytes that may no longer exist (hard retention mode purges blobs
|
||||
// for actively-referenced, potentially still-hot media), producing
|
||||
// an internal error instead of the graceful LOCATION_INVALID a stock
|
||||
// client knows how to render.
|
||||
s.blobCache.delete(b.LocationKey)
|
||||
refs, err := store.CountFileBlobRefs(ctx, string(b.Backend), b.ObjectKey)
|
||||
if err != nil {
|
||||
s.log.Warn("count file blob refs failed", zap.String("object_key", b.ObjectKey), zap.Error(err))
|
||||
|
|
@ -123,5 +202,6 @@ func (s *Service) deleteOrphanedBlobs(ctx context.Context, store mediaRetentionS
|
|||
if err := backend.Delete(ctx, b.ObjectKey); err != nil {
|
||||
s.log.Warn("delete orphaned blob failed", zap.String("object_key", b.ObjectKey), zap.Error(err))
|
||||
}
|
||||
s.byteCache.delete(b.ObjectKey)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -72,6 +72,10 @@ type Service struct {
|
|||
stickerSetNegCache *stickerSetNegativeCache
|
||||
uploadQuota domain.UploadPartQuota
|
||||
spaceGuard SpaceGuard
|
||||
// maxUploadFileBytes caps a single upload's total assembled size (sum of
|
||||
// all parts). <=0 means unlimited (still bounded by the protocol's own
|
||||
// MaxUploadPartBytes*MaxUploadParts ceiling). See WithMaxUploadFileBytes.
|
||||
maxUploadFileBytes int64
|
||||
mapTiles *mapTileProxy
|
||||
externalMedia *externalMediaFetcher
|
||||
webpage *webpageFetcher
|
||||
|
|
@ -143,6 +147,15 @@ func WithSpaceGuard(guard SpaceGuard) Option {
|
|||
}
|
||||
}
|
||||
|
||||
// WithMaxUploadFileBytes caps a single uploaded file's total assembled size
|
||||
// (sum of all parts, not any one part) -- TELESRV_STORAGE_MAX_UPLOAD_FILE_BYTES.
|
||||
// <=0 leaves it unlimited (still bounded by MaxUploadPartBytes*MaxUploadParts).
|
||||
func WithMaxUploadFileBytes(maxBytes int64) Option {
|
||||
return func(s *Service) {
|
||||
s.maxUploadFileBytes = maxBytes
|
||||
}
|
||||
}
|
||||
|
||||
// WithAdditionalBlobBackend registers a non-active blob backend purely for
|
||||
// reading/deleting rows written while it used to be active. Register the
|
||||
// previous backend here whenever TELESRV_BLOB_BACKEND changes and the old
|
||||
|
|
@ -753,6 +766,11 @@ func (s *Service) loadAndValidateUploadParts(ctx context.Context, ownerUserID, f
|
|||
return nil, 0, domain.ErrFilePartsInvalid
|
||||
}
|
||||
}
|
||||
// Checked against the total assembled size, not any single part --
|
||||
// MaxUploadPartBytes above already bounds each part individually.
|
||||
if s.maxUploadFileBytes > 0 && total > s.maxUploadFileBytes {
|
||||
return nil, 0, domain.ErrFileTooLarge
|
||||
}
|
||||
return parts, total, nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -171,6 +171,89 @@ func TestCreatePhotoFromUploadReceiptReplaysAfterPartCleanup(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// TestMaxUploadFileBytesRejectsOversizedAssembledDocument covers
|
||||
// TELESRV_STORAGE_MAX_UPLOAD_FILE_BYTES enforcement (WithMaxUploadFileBytes):
|
||||
// the check must fire against the TOTAL assembled size (sum of every part),
|
||||
// not any single part -- each individual part here is well under the limit,
|
||||
// only their sum exceeds it.
|
||||
func TestMaxUploadFileBytesRejectsOversizedAssembledDocument(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
media := newFakeMediaStore()
|
||||
local, err := NewLocalFS(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewLocalFS: %v", err)
|
||||
}
|
||||
const partSize = 1024
|
||||
const maxUploadFileBytes = 2 * partSize // exactly 2 parts' worth; 3 parts must be rejected
|
||||
svc := NewService(media, local, 2, WithVideoThumbnailer(nil), WithMaxUploadFileBytes(maxUploadFileBytes))
|
||||
|
||||
parts := []string{
|
||||
strings.Repeat("a", partSize),
|
||||
strings.Repeat("b", partSize),
|
||||
strings.Repeat("c", partSize),
|
||||
}
|
||||
for i, part := range parts {
|
||||
if _, err := svc.SaveBigFilePart(ctx, 10, 300, i, len(parts), []byte(part)); err != nil {
|
||||
t.Fatalf("SaveBigFilePart %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
_, err = svc.CreateDocumentFromUpload(ctx,
|
||||
domain.UploadedFileRef{OwnerUserID: 10, FileID: 300, Parts: len(parts), Name: "big.bin", Big: true},
|
||||
domain.DocumentSpec{MimeType: "application/octet-stream"},
|
||||
)
|
||||
if !errors.Is(err, domain.ErrFileTooLarge) {
|
||||
t.Fatalf("CreateDocumentFromUpload over max size err = %v, want ErrFileTooLarge", err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMaxUploadFileBytesAllowsAssembledSizeAtOrUnderLimit ensures the check
|
||||
// is an upper bound, not an off-by-one trap: a total exactly at the
|
||||
// configured ceiling must still succeed.
|
||||
func TestMaxUploadFileBytesAllowsAssembledSizeAtOrUnderLimit(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
media := newFakeMediaStore()
|
||||
local, err := NewLocalFS(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewLocalFS: %v", err)
|
||||
}
|
||||
const partSize = 1024
|
||||
const maxUploadFileBytes = 2 * partSize
|
||||
svc := NewService(media, local, 2, WithVideoThumbnailer(nil), WithMaxUploadFileBytes(maxUploadFileBytes))
|
||||
|
||||
parts := []string{strings.Repeat("a", partSize), strings.Repeat("b", partSize)}
|
||||
for i, part := range parts {
|
||||
if _, err := svc.SaveBigFilePart(ctx, 10, 301, i, len(parts), []byte(part)); err != nil {
|
||||
t.Fatalf("SaveBigFilePart %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
doc, err := svc.CreateDocumentFromUpload(ctx,
|
||||
domain.UploadedFileRef{OwnerUserID: 10, FileID: 301, Parts: len(parts), Name: "exact.bin", Big: true},
|
||||
domain.DocumentSpec{MimeType: "application/octet-stream"},
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("CreateDocumentFromUpload at exact max size: %v", err)
|
||||
}
|
||||
if doc.Size != int64(maxUploadFileBytes) {
|
||||
t.Fatalf("doc size = %d, want %d", doc.Size, maxUploadFileBytes)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMaxUploadFileBytesUnlimitedByDefault ensures leaving
|
||||
// WithMaxUploadFileBytes unset (or 0) never rejects on size -- only the
|
||||
// protocol's own MaxUploadPartBytes/MaxUploadParts ceiling still applies.
|
||||
func TestMaxUploadFileBytesUnlimitedByDefault(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
media := newFakeMediaStore()
|
||||
svc, _ := newUploadPartTestService(t, media, domain.UploadPartQuota{})
|
||||
file := domain.UploadedFileRef{OwnerUserID: 10, FileID: 302, Parts: 1, Name: "photo.jpg"}
|
||||
if _, err := svc.SaveFilePart(ctx, file.OwnerUserID, file.FileID, 0, []byte(strings.Repeat("z", 4096))); err != nil {
|
||||
t.Fatalf("SaveFilePart: %v", err)
|
||||
}
|
||||
if _, err := svc.CreatePhotoFromUpload(ctx, file); err != nil {
|
||||
t.Fatalf("CreatePhotoFromUpload with no configured max size: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
type countingUploadPartBackend struct {
|
||||
*LocalFS
|
||||
getUploadPartCalls int
|
||||
|
|
|
|||
|
|
@ -19,9 +19,38 @@ type OrphanedMediaRetentionStore interface {
|
|||
// how long a document/photo must have been orphaned before it's actually
|
||||
// deleted (not how old the media itself is) -- <=0 leaves the sweep
|
||||
// disabled even if a store is provided, matching
|
||||
// TELESRV_STORAGE_RETENTION_ENABLE=false being the safe default.
|
||||
// TELESRV_STORAGE_RETENTION_MODE=off/orphan-with-no-age being the safe
|
||||
// default. Mutually exclusive with WithHardMediaRetention -- the resolved
|
||||
// TELESRV_STORAGE_RETENTION_MODE is one of "off"/"orphan"/"hard", never
|
||||
// both orphan and hard sweeps at once.
|
||||
func (w *RetentionWorker) WithOrphanedMediaRetention(store OrphanedMediaRetentionStore, maxAge time.Duration) *RetentionWorker {
|
||||
w.orphanedMedia = store
|
||||
w.orphanedMediaMaxAge = maxAge
|
||||
return w
|
||||
}
|
||||
|
||||
// HardMediaRetentionStore physically deletes a document/photo's blob bytes
|
||||
// once the media itself (its upload/created time, not how long it has been
|
||||
// orphaned) is older than the configured age, REGARDLESS of whether a live
|
||||
// message/profile-photo/sticker-set still references it. Unlike
|
||||
// OrphanedMediaRetentionStore, it must never delete the document/photo
|
||||
// metadata row -- only the underlying file_blobs row(s) and, once confirming
|
||||
// no other row still needs the same content-addressed object, the physical
|
||||
// bytes. This keeps a message rendering its media placeholder (dimensions,
|
||||
// mime type, filename) after the bytes are gone; a subsequent download
|
||||
// resolves to LOCATION_INVALID instead of the message breaking outright.
|
||||
type HardMediaRetentionStore interface {
|
||||
DeleteBlobBytesForMediaOlderThan(ctx context.Context, cutoff time.Time, limit int) (int, error)
|
||||
}
|
||||
|
||||
// WithHardMediaRetention enables the aggressive storage retention sweep
|
||||
// (TELESRV_STORAGE_RETENTION_MODE=hard). maxAge is how old the media itself
|
||||
// must be (its created_at, not an orphaned_at grace period) before its blob
|
||||
// bytes are purged -- <=0 leaves the sweep disabled even if a store is
|
||||
// provided. Mutually exclusive with WithOrphanedMediaRetention -- call at
|
||||
// most one of the two, matching the config's single 3-way retention mode.
|
||||
func (w *RetentionWorker) WithHardMediaRetention(store HardMediaRetentionStore, maxAge time.Duration) *RetentionWorker {
|
||||
w.hardMedia = store
|
||||
w.hardMediaMaxAge = maxAge
|
||||
return w
|
||||
}
|
||||
|
|
|
|||
|
|
@ -95,6 +95,10 @@ const (
|
|||
// 继续保留，在线漏推由正常 difference 路径补偿。
|
||||
defaultOutboxPoisonRetention = time.Minute
|
||||
defaultOutboxPoisonInterval = 15 * time.Second
|
||||
// minMediaRetentionInterval floors the derived media-sweep cadence below
|
||||
// so a very short (e.g. test-only) TELESRV_STORAGE_RETENTION_MAX_AGE
|
||||
// can't spin the sweep query in a tight loop.
|
||||
minMediaRetentionInterval = 15 * time.Second
|
||||
)
|
||||
|
||||
// RetentionWorker 周期性回收存储中的死数据。
|
||||
|
|
@ -119,6 +123,7 @@ type RetentionWorker struct {
|
|||
activeAuthKeys ActiveRawAuthKeyProvider
|
||||
activeAuthKeyHeartbeat ActiveAuthKeyHeartbeatStore
|
||||
orphanedMedia OrphanedMediaRetentionStore
|
||||
hardMedia HardMediaRetentionStore
|
||||
logger *zap.Logger
|
||||
retention time.Duration
|
||||
botAPIRetention time.Duration
|
||||
|
|
@ -128,6 +133,7 @@ type RetentionWorker struct {
|
|||
outboxPoisonRetention time.Duration
|
||||
outboxPoisonInterval time.Duration
|
||||
orphanedMediaMaxAge time.Duration
|
||||
hardMediaMaxAge time.Duration
|
||||
interval time.Duration
|
||||
batch int
|
||||
}
|
||||
|
|
@ -259,6 +265,15 @@ func (w *RetentionWorker) Run(ctx context.Context) {
|
|||
heartbeatC = heartbeatTicker.C
|
||||
defer heartbeatTicker.Stop()
|
||||
}
|
||||
var (
|
||||
mediaTicker *time.Ticker
|
||||
mediaC <-chan time.Time
|
||||
)
|
||||
if interval := w.mediaRetentionInterval(); interval > 0 {
|
||||
mediaTicker = time.NewTicker(interval)
|
||||
mediaC = mediaTicker.C
|
||||
defer mediaTicker.Stop()
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
|
|
@ -269,6 +284,8 @@ func (w *RetentionWorker) Run(ctx context.Context) {
|
|||
w.runOutboxPoisonOnce(ctx)
|
||||
case <-heartbeatC:
|
||||
w.heartbeatActiveAuthKeys(ctx)
|
||||
case <-mediaC:
|
||||
w.runMediaRetentionOnce(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -276,6 +293,63 @@ func (w *RetentionWorker) Run(ctx context.Context) {
|
|||
func (w *RetentionWorker) runOnce(ctx context.Context) {
|
||||
w.runOutboxPoisonOnce(ctx)
|
||||
w.runRetentionOnce(ctx)
|
||||
w.runMediaRetentionOnce(ctx)
|
||||
}
|
||||
|
||||
// mediaRetentionInterval derives how often the storage media sweep
|
||||
// (orphan/hard) runs, independent of the shared housekeeping w.interval.
|
||||
// A short TELESRV_STORAGE_RETENTION_MAX_AGE (e.g. 30m) configured under a
|
||||
// much longer TELESRV_RETENTION_INTERVAL (default 1h) would otherwise let
|
||||
// media sit for up to age+interval past its cutoff before actually being
|
||||
// swept -- capping the sweep cadence at the retention age itself bounds
|
||||
// that worst case to at most 2x the configured age instead.
|
||||
func (w *RetentionWorker) mediaRetentionInterval() time.Duration {
|
||||
var maxAge time.Duration
|
||||
if w.orphanedMedia != nil && w.orphanedMediaMaxAge > 0 {
|
||||
maxAge = w.orphanedMediaMaxAge
|
||||
}
|
||||
if w.hardMedia != nil && w.hardMediaMaxAge > 0 && (maxAge == 0 || w.hardMediaMaxAge < maxAge) {
|
||||
maxAge = w.hardMediaMaxAge
|
||||
}
|
||||
if maxAge <= 0 {
|
||||
return 0
|
||||
}
|
||||
interval := w.interval
|
||||
if interval <= 0 || maxAge < interval {
|
||||
interval = maxAge
|
||||
}
|
||||
if interval < minMediaRetentionInterval {
|
||||
interval = minMediaRetentionInterval
|
||||
}
|
||||
return interval
|
||||
}
|
||||
|
||||
func (w *RetentionWorker) runMediaRetentionOnce(ctx context.Context) {
|
||||
if w.orphanedMedia != nil && w.orphanedMediaMaxAge > 0 {
|
||||
// Only ever touches documents/photos already marked orphaned (no live
|
||||
// message/profile-photo/sticker-set reference remains) -- media still
|
||||
// visible in a conversation is never a candidate, regardless of age.
|
||||
mediaDeleted, err := w.orphanedMedia.DeleteOrphanedOlderThan(ctx, time.Now().Add(-w.orphanedMediaMaxAge), w.batch)
|
||||
if err != nil {
|
||||
w.logger.Warn("orphaned media storage retention sweep failed", zap.Error(err))
|
||||
} else if mediaDeleted > 0 {
|
||||
w.logger.Info("orphaned media storage retention sweep complete", zap.Int("deleted", mediaDeleted))
|
||||
}
|
||||
}
|
||||
if w.hardMedia != nil && w.hardMediaMaxAge > 0 {
|
||||
// "Hard" mode: purges blob bytes for documents/photos older than
|
||||
// hardMediaMaxAge regardless of whether a live reference remains --
|
||||
// the store is required to keep the document/photo metadata row
|
||||
// intact, only removing file_blobs rows/bytes, so a message still
|
||||
// renders its media placeholder (LOCATION_INVALID on download)
|
||||
// instead of breaking outright.
|
||||
hardDeleted, err := w.hardMedia.DeleteBlobBytesForMediaOlderThan(ctx, time.Now().Add(-w.hardMediaMaxAge), w.batch)
|
||||
if err != nil {
|
||||
w.logger.Warn("hard media storage retention sweep failed", zap.Error(err))
|
||||
} else if hardDeleted > 0 {
|
||||
w.logger.Info("hard media storage retention sweep complete", zap.Int("deleted", hardDeleted))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *RetentionWorker) runOutboxPoisonOnce(ctx context.Context) {
|
||||
|
|
@ -414,17 +488,6 @@ func (w *RetentionWorker) runRetentionOnce(ctx context.Context) {
|
|||
w.logger.Info("expired channel_update_events contiguous-prefix cleanup complete", zap.Int("deleted", channelDeleted))
|
||||
}
|
||||
}
|
||||
if w.orphanedMedia != nil && w.orphanedMediaMaxAge > 0 {
|
||||
// Only ever touches documents/photos already marked orphaned (no live
|
||||
// message/profile-photo/sticker-set reference remains) -- media still
|
||||
// visible in a conversation is never a candidate, regardless of age.
|
||||
mediaDeleted, err := w.orphanedMedia.DeleteOrphanedOlderThan(ctx, time.Now().Add(-w.orphanedMediaMaxAge), w.batch)
|
||||
if err != nil {
|
||||
w.logger.Warn("orphaned media storage retention sweep failed", zap.Error(err))
|
||||
} else if mediaDeleted > 0 {
|
||||
w.logger.Info("orphaned media storage retention sweep complete", zap.Int("deleted", mediaDeleted))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *RetentionWorker) orphanHeartbeatInterval() time.Duration {
|
||||
|
|
|
|||
|
|
@ -393,3 +393,67 @@ func TestRetentionWorkerSkipsOrphanDeleteWhenHeartbeatFails(t *testing.T) {
|
|||
t.Fatalf("heartbeat failure signals = %+v", entries)
|
||||
}
|
||||
}
|
||||
|
||||
type fakeHardMediaRetention struct {
|
||||
calls int
|
||||
cutoff time.Time
|
||||
limit int
|
||||
deleted int
|
||||
}
|
||||
|
||||
func (f *fakeHardMediaRetention) DeleteBlobBytesForMediaOlderThan(_ context.Context, cutoff time.Time, limit int) (int, error) {
|
||||
f.calls++
|
||||
f.cutoff = cutoff
|
||||
f.limit = limit
|
||||
return f.deleted, nil
|
||||
}
|
||||
|
||||
// TestRetentionWorkerMediaSweepIntervalNeverExceedsMaxAge guards the fix for
|
||||
// a real gotcha reported live: with a short storage retention age (e.g. 30m)
|
||||
// under the default 1h TELESRV_RETENTION_INTERVAL housekeeping cadence,
|
||||
// media eligible for hard/orphan deletion used to sit for up to age+interval
|
||||
// before the shared ticker actually got around to sweeping it (a 30m age
|
||||
// could take up to 1h30m to actually disappear). The sweep now runs on its
|
||||
// own ticker capped at the configured age, so it never lags by more than
|
||||
// roughly one age-window.
|
||||
func TestRetentionWorkerMediaSweepIntervalNeverExceedsMaxAge(t *testing.T) {
|
||||
hard := &fakeHardMediaRetention{}
|
||||
w := NewRetentionWorker(&fakeOutboxRetention{}, nil, zap.NewNop(), time.Hour, time.Hour, 50).
|
||||
WithHardMediaRetention(hard, 30*time.Minute)
|
||||
|
||||
if got := w.mediaRetentionInterval(); got != 30*time.Minute {
|
||||
t.Fatalf("media retention interval = %v, want min(interval=1h, maxAge=30m) = 30m", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetentionWorkerMediaSweepIntervalFloorsAtMinimum(t *testing.T) {
|
||||
hard := &fakeHardMediaRetention{}
|
||||
w := NewRetentionWorker(&fakeOutboxRetention{}, nil, zap.NewNop(), time.Hour, time.Hour, 50).
|
||||
WithHardMediaRetention(hard, time.Second)
|
||||
|
||||
if got := w.mediaRetentionInterval(); got != minMediaRetentionInterval {
|
||||
t.Fatalf("media retention interval = %v, want floor %v", got, minMediaRetentionInterval)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetentionWorkerMediaSweepIntervalZeroWhenNoModeConfigured(t *testing.T) {
|
||||
w := NewRetentionWorker(&fakeOutboxRetention{}, nil, zap.NewNop(), time.Hour, time.Hour, 50)
|
||||
if got := w.mediaRetentionInterval(); got != 0 {
|
||||
t.Fatalf("media retention interval = %v, want 0 (no separate ticker) when retention is off", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRetentionWorkerRunsHardMediaSweepOnStartup confirms runOnce (called
|
||||
// once immediately when Run starts, matching every other retention check)
|
||||
// still fires the media sweep too, not just the new dedicated ticker.
|
||||
func TestRetentionWorkerRunsHardMediaSweepOnStartup(t *testing.T) {
|
||||
hard := &fakeHardMediaRetention{deleted: 4}
|
||||
w := NewRetentionWorker(&fakeOutboxRetention{}, nil, zap.NewNop(), time.Hour, time.Hour, 50).
|
||||
WithHardMediaRetention(hard, 30*time.Minute)
|
||||
|
||||
w.runOnce(context.Background())
|
||||
|
||||
if hard.calls != 1 {
|
||||
t.Fatalf("hard media sweep calls = %d, want 1", hard.calls)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue