fix(photos): faster large-album loading and album cover lifecycle

Large albums opened slowly and flickered; the default album cover was
also lost in the media-objects refactor.

Large-album loading:
- quantize signed object URLs within a window and send Cache-Control so
  previews are reused instead of re-fetched
- publish photo.updated (not photo.created) when variants are ready, so
  clients update in place without reordering the grid
- worker concurrency + newest-first variant priority; raise default batch

Album cover:
- first uploaded photo becomes the cover when none is set
- clear the cover when the cover photo is trashed or moved out
- "set as album cover" in the photo context menu and actions panel

OpenSpec changes: fix-large-album-load, fix-album-cover
This commit is contained in:
iu
2026-10-11 14:15:59 +03:00
parent 6361e48b20
commit d7a37328f0
40 changed files with 999 additions and 57 deletions
+18 -1
View File
@@ -3,6 +3,7 @@ package app
import (
"context"
"errors"
"fmt"
"io/fs"
"net/http"
"strings"
@@ -59,7 +60,7 @@ func New(cfg *config.Config, repos store.Repos, ping func(ctx context.Context) e
membersSvc := members.NewService(repos.Members, repos.Albums, hub, notifSvc)
photosSvc := photos.NewService(repos.Photos, albumsSvc, st, hub)
photosWorker := photos.NewWorker(photosSvc, cfg.Worker.Interval, cfg.Worker.Batch, cfg.Worker.PendingTTL)
photosWorker := photos.NewWorker(photosSvc, cfg.Worker.Interval, cfg.Worker.Batch, cfg.Worker.PendingTTL, cfg.Worker.Concurrency)
socialSvc := social.NewService(repos.Social, albumsSvc, repos.Photos, hub, notifSvc)
@@ -198,20 +199,36 @@ func (s *Server) health(c *gin.Context) {
func (s *Server) serveObject(c *gin.Context) {
key := strings.TrimPrefix(c.Param("key"), "/")
// Objects are immutable (one key maps to one blob), so successful responses
// are safe to cache. The header is dropped on error paths below.
c.Writer.Header().Set("Cache-Control", cacheControlValue(s.cfg.Storage.CacheMaxAge))
switch err := s.storage.ServeObject(c.Writer, c.Request, key); {
case err == nil:
return
case errors.Is(err, storage.ErrSignedURLInvalid):
c.Writer.Header().Del("Cache-Control")
c.Status(http.StatusForbidden)
case errors.Is(err, storage.ErrObjectNotFound):
c.Writer.Header().Del("Cache-Control")
httpx.RespondError(c, httpx.NotFound("object_not_found", "object not found"))
case errors.Is(err, storage.ErrRangeNotSatisfiable):
c.Writer.Header().Del("Cache-Control")
c.Status(http.StatusRequestedRangeNotSatisfiable)
default:
c.Writer.Header().Del("Cache-Control")
c.Status(http.StatusInternalServerError)
}
}
// cacheControlValue renders the Cache-Control value for object responses.
// Non-positive values fall back to a one-hour freshness window.
func cacheControlValue(maxAge time.Duration) string {
if maxAge <= 0 {
maxAge = time.Hour
}
return fmt.Sprintf("private, max-age=%d", int(maxAge.Seconds()))
}
func (s *Server) albumSSE(c *gin.Context) {
userID := auth.UserID(c)
albumID := c.Param("id")
+1 -1
View File
@@ -16,7 +16,7 @@ import (
func mustSigner(t *testing.T) *storage.Signer {
t.Helper()
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("signer: %v", err)
}
+7 -1
View File
@@ -22,7 +22,7 @@ import (
func newTestServerWithFS(t *testing.T) (*storage.FS, *gin.Engine) {
t.Helper()
gin.SetMode(gin.TestMode)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("signer: %v", err)
}
@@ -70,6 +70,9 @@ func TestServeObjectRoute(t *testing.T) {
if rec.Body.String() != "0123456789" {
t.Fatalf("body = %q", rec.Body.String())
}
if cc := rec.Header().Get("Cache-Control"); cc == "" || !strings.Contains(cc, "max-age=") {
t.Fatalf("cache-control = %q, want max-age directive", cc)
}
})
t.Run("range", func(t *testing.T) {
@@ -89,6 +92,9 @@ func TestServeObjectRoute(t *testing.T) {
if rec.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403", rec.Code)
}
if cc := rec.Header().Get("Cache-Control"); cc != "" {
t.Fatalf("cache-control = %q, want empty on error", cc)
}
})
t.Run("forged", func(t *testing.T) {
+2 -2
View File
@@ -149,7 +149,7 @@ func runShareFlow(t *testing.T, backend storage.Backend, verify func(t *testing.
}
func TestShareFlowWithFS(t *testing.T) {
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("signer: %v", err)
}
@@ -178,7 +178,7 @@ func TestShareFlowWithS3(t *testing.T) {
if endpoint == "" {
t.Skip("S3_ENDPOINT not set")
}
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Skipf("signer init: %v", err)
}
+36 -15
View File
@@ -48,16 +48,19 @@ type SSE struct {
}
type Worker struct {
Interval time.Duration
Batch int
PendingTTL time.Duration
Interval time.Duration
Batch int
PendingTTL time.Duration
Concurrency int
}
type Storage struct {
Backend string
FSRoot string
URLSecret string
URLTTL time.Duration
Backend string
FSRoot string
URLSecret string
URLTTL time.Duration
URLSignWindow time.Duration
CacheMaxAge time.Duration
}
type Database struct {
@@ -103,6 +106,16 @@ func Load() (*Config, error) {
return nil, fmt.Errorf("invalid OBJECT_URL_TTL: %w", err)
}
urlSignWindow, err := time.ParseDuration(getEnv("OBJECT_URL_SIGN_WINDOW", "10m"))
if err != nil {
return nil, fmt.Errorf("invalid OBJECT_URL_SIGN_WINDOW: %w", err)
}
cacheMaxAge, err := time.ParseDuration(getEnv("OBJECT_CACHE_MAX_AGE", "1h"))
if err != nil {
return nil, fmt.Errorf("invalid OBJECT_CACHE_MAX_AGE: %w", err)
}
heartbeat, err := time.ParseDuration(getEnv("SSE_HEARTBEAT_INTERVAL", "25s"))
if err != nil {
return nil, fmt.Errorf("invalid SSE_HEARTBEAT_INTERVAL: %w", err)
@@ -123,7 +136,7 @@ func Load() (*Config, error) {
return nil, fmt.Errorf("invalid WORKER_INTERVAL: %w", err)
}
workerBatch, err := strconv.Atoi(getEnv("WORKER_BATCH", "10"))
workerBatch, err := strconv.Atoi(getEnv("WORKER_BATCH", "100"))
if err != nil {
return nil, fmt.Errorf("invalid WORKER_BATCH: %w", err)
}
@@ -133,6 +146,11 @@ func Load() (*Config, error) {
return nil, fmt.Errorf("invalid WORKER_PENDING_TTL: %w", err)
}
workerConcurrency, err := strconv.Atoi(getEnv("WORKER_CONCURRENCY", "4"))
if err != nil {
return nil, fmt.Errorf("invalid WORKER_CONCURRENCY: %w", err)
}
logLevel := getEnv("LOG_LEVEL", "info")
if _, err := logging.ParseLevel(logLevel); err != nil {
return nil, err
@@ -160,10 +178,12 @@ func Load() (*Config, error) {
UsePathStyle: pathStyle,
},
Storage: Storage{
Backend: getEnv("STORAGE_BACKEND", "fs"),
FSRoot: getEnv("FS_ROOT", "./data"),
URLSecret: getEnv("URL_SECRET", ""),
URLTTL: urlTTL,
Backend: getEnv("STORAGE_BACKEND", "fs"),
FSRoot: getEnv("FS_ROOT", "./data"),
URLSecret: getEnv("URL_SECRET", ""),
URLTTL: urlTTL,
URLSignWindow: urlSignWindow,
CacheMaxAge: cacheMaxAge,
},
Database: Database{
Backend: getEnv("DB_BACKEND", "sqlite"),
@@ -182,9 +202,10 @@ func Load() (*Config, error) {
ClientBuffer: bufSize,
},
Worker: Worker{
Interval: workerInterval,
Batch: workerBatch,
PendingTTL: workerPendingTTL,
Interval: workerInterval,
Batch: workerBatch,
PendingTTL: workerPendingTTL,
Concurrency: workerConcurrency,
},
Log: Log{
Level: logLevel,
+143 -4
View File
@@ -43,6 +43,7 @@ type stack struct {
photos *photos.Service
albums *albums.Service
auth *auth.Service
hub *sse.Hub
}
// openStack opens a migrated SQLite database at path and builds the services.
@@ -61,7 +62,7 @@ func openStack(t *testing.T, path, blobRoot string) (*stack, func()) {
}
repos := store.NewSQLite(database)
hub := sse.NewHub(1)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("signer: %v", err)
}
@@ -72,7 +73,145 @@ func openStack(t *testing.T, path, blobRoot string) (*stack, func()) {
authSvc := auth.NewService(repos.Auth, auth.NewTokenManager("secret", 15*time.Minute, time.Hour))
albumsSvc := albums.NewService(repos.Albums, repos.Members, hub, blob)
photosSvc := photos.NewService(repos.Photos, albumsSvc, blob, hub)
return &stack{db: database, photos: photosSvc, albums: albumsSvc, auth: authSvc}, func() { database.Close() }
return &stack{db: database, photos: photosSvc, albums: albumsSvc, auth: authSvc, hub: hub}, func() { database.Close() }
}
func TestUploadSetsDefaultCover(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "galo.db")
st, closeDB := openStack(t, path, t.TempDir())
defer closeDB()
user, _, err := st.auth.Register(ctx, "cover@test.local", "password123", "C")
if err != nil {
t.Fatalf("register: %v", err)
}
album, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "Cover"})
if err != nil {
t.Fatalf("create album: %v", err)
}
first, err := st.photos.Upload(ctx, photos.UploadInput{
AlbumID: album.ID, UserID: user.ID, Filename: "a.png", Data: makePNG(t),
})
if err != nil {
t.Fatalf("upload first: %v", err)
}
got, _, err := st.albums.Get(ctx, album.ID, user.ID)
if err != nil {
t.Fatalf("get album: %v", err)
}
if got.CoverPhotoID == nil || *got.CoverPhotoID != first.ID {
t.Fatalf("cover = %v, want first photo %s", got.CoverPhotoID, first.ID)
}
second, err := st.photos.Upload(ctx, photos.UploadInput{
AlbumID: album.ID, UserID: user.ID, Filename: "b.png", Data: makePNG(t),
})
if err != nil {
t.Fatalf("upload second: %v", err)
}
got, _, err = st.albums.Get(ctx, album.ID, user.ID)
if err != nil {
t.Fatalf("get album after second: %v", err)
}
if got.CoverPhotoID == nil || *got.CoverPhotoID != first.ID {
t.Fatalf("cover changed to %v, want still first %s (second=%s)", got.CoverPhotoID, first.ID, second.ID)
}
}
func TestCoverClearedWhenCoverPhotoLeaves(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "galo.db")
st, closeDB := openStack(t, path, t.TempDir())
defer closeDB()
user, _, err := st.auth.Register(ctx, "clear@test.local", "password123", "C")
if err != nil {
t.Fatalf("register: %v", err)
}
albumA, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "A"})
if err != nil {
t.Fatalf("create A: %v", err)
}
albumB, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "B"})
if err != nil {
t.Fatalf("create B: %v", err)
}
cover, err := st.photos.Upload(ctx, photos.UploadInput{
AlbumID: albumA.ID, UserID: user.ID, Filename: "cover.png", Data: makePNG(t),
})
if err != nil {
t.Fatalf("upload cover: %v", err)
}
// Trashing the cover clears it.
if err := st.photos.Delete(ctx, cover.ID, user.ID); err != nil {
t.Fatalf("delete cover: %v", err)
}
goal, _, _ := st.albums.Get(ctx, albumA.ID, user.ID)
if goal.CoverPhotoID != nil {
t.Fatalf("cover not cleared on delete: %v", goal.CoverPhotoID)
}
// A fresh upload becomes the cover, and moving it out clears it again.
next, err := st.photos.Upload(ctx, photos.UploadInput{
AlbumID: albumA.ID, UserID: user.ID, Filename: "next.png", Data: makePNG(t),
})
if err != nil {
t.Fatalf("upload next: %v", err)
}
goal, _, _ = st.albums.Get(ctx, albumA.ID, user.ID)
if goal.CoverPhotoID == nil || *goal.CoverPhotoID != next.ID {
t.Fatalf("cover = %v, want %s", goal.CoverPhotoID, next.ID)
}
if _, err := st.photos.MoveBatch(ctx, []string{next.ID}, user.ID, albumB.ID); err != nil {
t.Fatalf("move: %v", err)
}
goal, _, _ = st.albums.Get(ctx, albumA.ID, user.ID)
if goal.CoverPhotoID != nil {
t.Fatalf("cover not cleared on move: %v", goal.CoverPhotoID)
}
}
func TestWorkerPublishesPhotoUpdated(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "galo.db")
st, closeDB := openStack(t, path, t.TempDir())
defer closeDB()
worker := photos.NewWorker(st.photos, 20*time.Millisecond, 10, time.Minute, 2)
wctx, cancel := context.WithCancel(ctx)
defer cancel()
go worker.Run(wctx)
user, _, err := st.auth.Register(ctx, "evt@test.local", "password123", "E")
if err != nil {
t.Fatalf("register: %v", err)
}
album, err := st.albums.Create(ctx, user.ID, albums.CreateInput{Title: "Trip"})
if err != nil {
t.Fatalf("create album: %v", err)
}
sub := st.hub.SubscribeAlbum(album.ID)
defer sub.Close()
if _, err := st.photos.Upload(ctx, photos.UploadInput{
AlbumID: album.ID, UserID: user.ID, Filename: "a.png", Data: makePNG(t),
}); err != nil {
t.Fatalf("upload: %v", err)
}
select {
case ev := <-sub.Events():
if ev.Type != "photo.updated" {
t.Fatalf("variant-ready event = %q, want photo.updated", ev.Type)
}
case <-time.After(5 * time.Second):
t.Fatalf("no photo.updated event published for generated variants")
}
}
func TestSQLiteEndToEndWithRestart(t *testing.T) {
@@ -81,7 +220,7 @@ func TestSQLiteEndToEndWithRestart(t *testing.T) {
blobs := t.TempDir()
st, closeDB := openStack(t, path, blobs)
worker := photos.NewWorker(st.photos, 20*time.Millisecond, 10, time.Minute)
worker := photos.NewWorker(st.photos, 20*time.Millisecond, 10, time.Minute, 2)
wctx, cancel := context.WithCancel(ctx)
go worker.Run(wctx)
@@ -127,7 +266,7 @@ func TestSQLiteEndToEndWithRestart(t *testing.T) {
st2, close2 := openStack(t, path, blobs)
defer close2()
worker2 := photos.NewWorker(st2.photos, 20*time.Millisecond, 10, time.Minute)
worker2 := photos.NewWorker(st2.photos, 20*time.Millisecond, 10, time.Minute, 2)
wctx2, cancel2 := context.WithCancel(ctx)
defer cancel2()
go worker2.Run(wctx2)
+1 -1
View File
@@ -22,7 +22,7 @@ func TestCopySharesStorageObjects(t *testing.T) {
}
t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, p.ID) })
w := NewWorker(svc, time.Second, 10, time.Minute)
w := NewWorker(svc, time.Second, 10, time.Minute, 2)
w.processVariantJobs(ctx)
srcIDs, err := svc.store.PhotoObjectIDs(ctx, p.ID)
+2 -2
View File
@@ -106,7 +106,7 @@ func newTestService(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend
if endpoint == "" {
t.Skip("S3_ENDPOINT not set")
}
signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour, 10*time.Minute)
if err != nil {
t.Skipf("signer init: %v", err)
}
@@ -128,7 +128,7 @@ func newTestService(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend
// storage backend so the full flow can run without external infrastructure.
func newTestServiceFS(t *testing.T, pool *pgxpool.Pool) (*Service, storage.Backend) {
t.Helper()
signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour)
signer, err := storage.NewSigner("http://app.test", "test-secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("storage init: %v", err)
}
+8 -4
View File
@@ -296,11 +296,15 @@ func (s *Store) ClaimVariantJobs(ctx context.Context, limit int, lease time.Dura
defer func() { _ = tx.Rollback(ctx) }()
rows, err := tx.Query(ctx,
`SELECT photo_id FROM variant_jobs
WHERE next_attempt_at <= now()
ORDER BY next_attempt_at
`SELECT vj.photo_id FROM variant_jobs vj
JOIN photos p ON p.id = vj.photo_id
WHERE vj.next_attempt_at <= now()
ORDER BY (vj.attempts > 0) ASC,
CASE WHEN vj.attempts = 0 THEN p.created_at END DESC NULLS LAST,
vj.next_attempt_at ASC,
vj.photo_id DESC
LIMIT $1
FOR UPDATE SKIP LOCKED`, limit)
FOR UPDATE OF vj SKIP LOCKED`, limit)
if err != nil {
return nil, err
}
+1
View File
@@ -28,6 +28,7 @@ type Repository interface {
DetachTag(ctx context.Context, photoID, tagID string) error
TagOwnedBy(ctx context.Context, tagID, userID string) (bool, error)
SetCoverIfEmpty(ctx context.Context, albumID, photoID string)
ClearCoverIfMatches(ctx context.Context, albumID, photoID string)
UpdateFilename(ctx context.Context, id, filename string) error
BackfillEXIF(ctx context.Context, fetch func(context.Context, string) ([]byte, error)) (int, error)
+7
View File
@@ -110,6 +110,9 @@ func (s *Service) Upload(ctx context.Context, in UploadInput) (*models.Photo, er
return nil, err
}
// The first photo uploaded to an album becomes its default cover.
s.store.SetCoverIfEmpty(ctx, in.AlbumID, p.ID)
if err := s.attachURLs(ctx, p); err != nil {
return nil, err
}
@@ -298,6 +301,8 @@ func (s *Service) Delete(ctx context.Context, photoID, userID string) error {
if err := s.store.SetDeleted(ctx, photoID, true); err != nil {
return err
}
// A trashed cover leaves the album without one.
s.store.ClearCoverIfMatches(ctx, p.AlbumID, photoID)
s.publishDeleted(p)
return nil
}
@@ -451,6 +456,8 @@ func (s *Service) MoveBatch(ctx context.Context, ids []string, userID, targetAlb
if err := s.store.UpdateAlbum(ctx, id, targetAlbumID); err != nil {
continue
}
// Moving a cover photo out of its album leaves the album without one.
s.store.ClearCoverIfMatches(ctx, oldAlbum, id)
p.AlbumID = targetAlbumID
s.hub.Publish(oldAlbum, "photo.deleted", map[string]string{
"id": p.ID, "album_id": oldAlbum, "uploader_id": p.UploaderID,
+7 -3
View File
@@ -240,9 +240,13 @@ func (s *SQLiteStore) ClaimVariantJobs(ctx context.Context, limit int, lease tim
defer func() { _ = tx.Rollback() }()
rows, err := tx.QueryContext(ctx,
`SELECT photo_id FROM variant_jobs
WHERE next_attempt_at <= $1
ORDER BY next_attempt_at
`SELECT vj.photo_id FROM variant_jobs vj
JOIN photos p ON p.id = vj.photo_id
WHERE vj.next_attempt_at <= $1
ORDER BY (vj.attempts > 0) ASC,
CASE WHEN vj.attempts = 0 THEN p.created_at END DESC NULLS LAST,
vj.next_attempt_at ASC,
vj.photo_id DESC
LIMIT $2`, now, limit)
if err != nil {
return nil, err
+8
View File
@@ -403,6 +403,14 @@ func (s *SQLiteStore) SetCoverIfEmpty(ctx context.Context, albumID, photoID stri
albumID, photoID)
}
// ClearCoverIfMatches removes the album cover only when it currently points at
// photoID, so a concurrent cover change is not clobbered.
func (s *SQLiteStore) ClearCoverIfMatches(ctx context.Context, albumID, photoID string) {
_, _ = s.db.ExecContext(ctx,
`UPDATE albums SET cover_photo_id = NULL WHERE id = $1 AND cover_photo_id = $2`,
albumID, photoID)
}
func (s *SQLiteStore) UpdateFilename(ctx context.Context, id, filename string) error {
res, err := s.db.ExecContext(ctx, `UPDATE photos SET original_filename = $2 WHERE id = $1`, id, filename)
if err != nil {
+8
View File
@@ -424,6 +424,14 @@ func (s *Store) SetCoverIfEmpty(ctx context.Context, albumID, photoID string) {
albumID, photoID)
}
// ClearCoverIfMatches removes the album cover only when it currently points at
// photoID, so a concurrent cover change is not clobbered.
func (s *Store) ClearCoverIfMatches(ctx context.Context, albumID, photoID string) {
_, _ = s.pool.Exec(ctx,
`UPDATE albums SET cover_photo_id = NULL WHERE id = $1 AND cover_photo_id = $2`,
albumID, photoID)
}
func (s *Store) UpdateFilename(ctx context.Context, id, filename string) error {
tag, err := s.pool.Exec(ctx, `UPDATE photos SET original_filename = $2 WHERE id = $1`, id, filename)
if err != nil {
+48 -13
View File
@@ -5,6 +5,7 @@ import (
"context"
"errors"
"log/slog"
"sync"
"time"
"galo/internal/auth"
@@ -16,15 +17,16 @@ const maxVariantAttempts = 5
// and the object garbage collector entirely from the state recorded in the
// metadata layer, so it resumes correctly after a restart.
type Worker struct {
svc *Service
interval time.Duration
batch int
pendingTTL time.Duration
lease time.Duration
svc *Service
interval time.Duration
batch int
pendingTTL time.Duration
lease time.Duration
concurrency int
}
// NewWorker builds a worker. Parameters are clamped to sane defaults.
func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration) *Worker {
func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.Duration, concurrency int) *Worker {
if interval <= 0 {
interval = 5 * time.Second
}
@@ -34,12 +36,16 @@ func NewWorker(svc *Service, interval time.Duration, batch int, pendingTTL time.
if pendingTTL <= 0 {
pendingTTL = 10 * time.Minute
}
if concurrency <= 0 {
concurrency = 4
}
return &Worker{
svc: svc,
interval: interval,
batch: batch,
pendingTTL: pendingTTL,
lease: interval * 4,
svc: svc,
interval: interval,
batch: batch,
pendingTTL: pendingTTL,
lease: interval * 4,
concurrency: concurrency,
}
}
@@ -71,9 +77,36 @@ func (w *Worker) processVariantJobs(ctx context.Context) {
slog.Error("claim variant jobs", "error", err)
return
}
for _, id := range ids {
w.forEachConcurrent(ids, func(id string) {
w.generateVariants(ctx, id)
})
}
// forEachConcurrent runs fn over ids with at most w.concurrency functions in
// flight. Generation is CPU-bound (decode + resize + encode), so parallelism
// scales with cores; a concurrency of 1 stays strictly sequential.
func (w *Worker) forEachConcurrent(ids []string, fn func(string)) {
if len(ids) == 0 {
return
}
if w.concurrency <= 1 {
for _, id := range ids {
fn(id)
}
return
}
sem := make(chan struct{}, w.concurrency)
var wg sync.WaitGroup
for _, id := range ids {
wg.Add(1)
sem <- struct{}{}
go func(id string) {
defer wg.Done()
defer func() { <-sem }()
fn(id)
}(id)
}
wg.Wait()
}
func (w *Worker) generateVariants(ctx context.Context, photoID string) {
@@ -162,7 +195,9 @@ func (w *Worker) notify(ctx context.Context, photoID string) {
if err := w.svc.attachURLs(ctx, p); err != nil {
return
}
w.svc.hub.Publish(p.AlbumID, "photo.created", p)
// Variant readiness updates an existing photo; it is not a new photo in the
// album, so clients update it in place instead of inserting at the top.
w.svc.hub.Publish(p.AlbumID, "photo.updated", p)
}
func (w *Worker) fail(ctx context.Context, photoID, msg string) {
+59 -1
View File
@@ -40,9 +40,21 @@ func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Ba
}
t.Cleanup(func() { cleanupPhoto(t, ctx, pool, cl, p.ID) })
w := NewWorker(svc, time.Second, 10, time.Minute)
sub := svc.hub.SubscribeAlbum(album)
defer sub.Close()
w := NewWorker(svc, time.Second, 10, time.Minute, 2)
w.processVariantJobs(ctx)
select {
case ev := <-sub.Events():
if ev.Type != "photo.updated" {
t.Fatalf("variant-ready event = %q, want photo.updated", ev.Type)
}
case <-time.After(time.Second):
t.Fatalf("no variant-ready event published")
}
refreshed, err := svc.store.GetByID(ctx, p.ID, "")
if err != nil {
t.Fatalf("get refreshed: %v", err)
@@ -82,6 +94,52 @@ func runWorkerFlow(t *testing.T, pool *pgxpool.Pool, svc *Service, cl storage.Ba
}
}
func TestWorkerConcurrencyPool(t *testing.T) {
w := &Worker{concurrency: 3}
ids := []string{"a", "b", "c", "d", "e"}
started := make(chan struct{}, len(ids))
release := make(chan struct{})
done := make(chan struct{})
go func() {
w.forEachConcurrent(ids, func(string) {
started <- struct{}{}
<-release
})
close(done)
}()
for i := 0; i < 3; i++ {
select {
case <-started:
case <-time.After(2 * time.Second):
t.Fatalf("only %d tasks started concurrently, want 3", i)
}
}
select {
case <-started:
t.Fatalf("more than concurrency tasks started concurrently")
case <-time.After(50 * time.Millisecond):
}
close(release)
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatalf("pool did not finish")
}
}
func TestWorkerConcurrencySequentialAtOne(t *testing.T) {
w := &Worker{concurrency: 1}
var order []string
w.forEachConcurrent([]string{"a", "b", "c"}, func(id string) {
order = append(order, id)
})
if len(order) != 3 || order[0] != "a" || order[1] != "b" || order[2] != "c" {
t.Fatalf("sequential order = %v, want [a b c]", order)
}
}
func countObjects(t *testing.T, ctx context.Context, pool *pgxpool.Pool, ids []string) int {
t.Helper()
var n int
@@ -382,6 +382,18 @@ func testPhotos(t *testing.T, ctx context.Context, r Repos) {
t.Fatalf("cover not set: %+v", al.CoverPhotoID)
}
// Clearing the cover only affects a matching photo id.
r.Photos.ClearCoverIfMatches(ctx, a.ID, newID())
al, _ = r.Albums.GetByID(ctx, a.ID)
if al.CoverPhotoID == nil || *al.CoverPhotoID != p.ID {
t.Fatalf("cover cleared by non-matching id: %+v", al.CoverPhotoID)
}
r.Photos.ClearCoverIfMatches(ctx, a.ID, p.ID)
al, _ = r.Albums.GetByID(ctx, a.ID)
if al.CoverPhotoID != nil {
t.Fatalf("cover not cleared: %+v", al.CoverPhotoID)
}
if err := r.Photos.SetDeleted(ctx, p.ID, true); err != nil {
t.Fatalf("soft delete: %v", err)
}
@@ -524,6 +536,19 @@ func testObjects(t *testing.T, ctx context.Context, r Repos) {
if jobs, _ = r.Photos.ClaimVariantJobs(ctx, 10, time.Minute); contains(jobs, p3.ID) {
t.Fatalf("abandoned job still present: %+v", jobs)
}
// Fresh jobs are prioritized by photo freshness: the newest photo's job is
// claimed first. The delay keeps the two photos' timestamps distinct.
pOlder := seedPhoto(t, ctx, r, owner.ID, a.ID)
time.Sleep(5 * time.Millisecond)
pNewer := seedPhoto(t, ctx, r, owner.ID, a.ID)
order, err := r.Photos.ClaimVariantJobs(ctx, 2, time.Minute)
if err != nil {
t.Fatalf("claim ordered jobs: %v", err)
}
if len(order) < 2 || order[0] != pNewer.ID || order[1] != pOlder.ID {
t.Fatalf("claim order = %+v, want newest first [%s %s]", order, pNewer.ID, pOlder.ID)
}
}
func testSettings(t *testing.T, ctx context.Context, r Repos) {
+1 -1
View File
@@ -16,7 +16,7 @@ import (
func newTestFS(t *testing.T) *FS {
t.Helper()
signer, err := NewSigner("http://app.test", "test-secret", time.Hour)
signer, err := NewSigner("http://app.test", "test-secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("signer: %v", err)
}
+1 -1
View File
@@ -22,7 +22,7 @@ func newTestS3(t *testing.T) (*S3, *Signer) {
if endpoint == "" {
t.Skip("S3_ENDPOINT not set")
}
signer, err := NewSigner("http://app.test", "secret", time.Hour)
signer, err := NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("signer: %v", err)
}
+30 -4
View File
@@ -20,33 +20,59 @@ type Signer struct {
secret []byte
baseURL string
ttl time.Duration
window time.Duration
now func() time.Time
}
// NewSigner builds a signer. secret signs the URLs (required); baseURL is the
// public base URL of the app; ttl limits URL lifetime (defaults to one hour).
func NewSigner(baseURL, secret string, ttl time.Duration) (*Signer, error) {
// public base URL of the app; ttl limits URL lifetime (defaults to one hour);
// window quantizes the expiry so the same object yields an identical URL within
// one window (defaults to ten minutes, and is capped at ttl). A non-positive
// window keeps the URL stable for its whole lifetime.
func NewSigner(baseURL, secret string, ttl, window time.Duration) (*Signer, error) {
if strings.TrimSpace(secret) == "" {
return nil, errors.New("url secret is required")
}
if ttl <= 0 {
ttl = time.Hour
}
if window <= 0 {
window = 10 * time.Minute
}
if window > ttl {
window = ttl
}
return &Signer{
secret: []byte(secret),
baseURL: strings.TrimRight(baseURL, "/"),
ttl: ttl,
window: window,
now: time.Now,
}, nil
}
// URL returns a signed app URL for the given opaque key.
// URL returns a signed app URL for the given opaque key. The expiry is
// quantized to the signing window so repeated calls within that window return
// the identical URL, letting clients reuse cached objects.
func (s *Signer) URL(key string) string {
exp := time.Now().Add(s.ttl).Unix()
exp := s.expiry()
q := url.Values{}
q.Set("e", strconv.FormatInt(exp, 10))
q.Set("s", s.sign(key, exp))
return fmt.Sprintf("%s/api/objects/%s?%s", s.baseURL, escapeKey(key), q.Encode())
}
// expiry is the quantized Unix expiry shared by all URLs issued in the current
// window: the window start plus the TTL. It is never earlier than now.
func (s *Signer) expiry() int64 {
now := s.now()
exp := now.Truncate(s.window).Add(s.ttl)
if exp.Before(now) {
exp = now.Add(s.ttl)
}
return exp.Unix()
}
// VerifyRequest checks the signature and expiry carried by an object request.
// It returns ErrSignedURLInvalid when the request is not authorized.
func (s *Signer) VerifyRequest(key string, r *http.Request) error {
+63
View File
@@ -0,0 +1,63 @@
package storage
import (
"net/http"
"net/http/httptest"
"net/url"
"strconv"
"testing"
"time"
)
func expiryOf(t *testing.T, raw string) int64 {
t.Helper()
u, err := url.Parse(raw)
if err != nil {
t.Fatalf("parse url: %v", err)
}
e, err := strconv.ParseInt(u.Query().Get("e"), 10, 64)
if err != nil {
t.Fatalf("parse expiry: %v", err)
}
return e
}
func TestSignerURLStableWithinWindow(t *testing.T) {
s, err := NewSigner("http://app.test", "secret", time.Hour, time.Hour)
if err != nil {
t.Fatalf("new signer: %v", err)
}
first := s.URL("objects/a/thumb.webp")
second := s.URL("objects/a/thumb.webp")
if first != second {
t.Fatalf("URL changed within window:\n %q\n %q", first, second)
}
}
func TestSignerURLChangesAcrossWindow(t *testing.T) {
s, err := NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("new signer: %v", err)
}
base := time.Now()
s.now = func() time.Time { return base }
first := expiryOf(t, s.URL("objects/a/thumb.webp"))
s.now = func() time.Time { return base.Add(s.window) }
second := expiryOf(t, s.URL("objects/a/thumb.webp"))
if first == second {
t.Fatalf("expiry did not advance across window: %d", first)
}
}
func TestSignerURLVerifies(t *testing.T) {
s, err := NewSigner("http://app.test", "secret", time.Hour, 10*time.Minute)
if err != nil {
t.Fatalf("new signer: %v", err)
}
key := "objects/a/thumb.webp"
raw := s.URL(key)
req := httptest.NewRequest(http.MethodGet, raw, nil)
if err := s.VerifyRequest(key, req); err != nil {
t.Fatalf("verify issued URL: %v", err)
}
}
+1 -1
View File
@@ -108,7 +108,7 @@ func Open(ctx context.Context, cfg *config.Config, jwtSecret string) (Backend, e
if secret == "" {
return nil, errors.New("object URL secret is not available")
}
signer, err := NewSigner(cfg.Server.BaseURL, secret, cfg.Storage.URLTTL)
signer, err := NewSigner(cfg.Server.BaseURL, secret, cfg.Storage.URLTTL, cfg.Storage.URLSignWindow)
if err != nil {
return nil, err
}