fix(api): handle expired asset metadata in AssetRefWriter (#469)

* fix(api): handle expired asset metadata in AssetRefWriter to prevent infinite retry loops

The flushProject function treated all Redis errors identically, restoring
pending counts for retry. When metadata keys expired (redis.Nil), this
created an infinite error loop since the metadata would never reappear.

Two fixes:
1. Enqueue now refreshes the meta key TTL on every call, preventing
   premature expiration while pending counts are still accumulating.
2. flushProject distinguishes redis.Nil (drop orphaned entry) from
   transient errors (restore for retry).

* fix(ci): prevent E2E test hanging by adding teardown and flush context

The E2E workflow hangs after tests pass because docker compose
waits for Postgres to stop, but dangling connections prevent clean
shutdown. Add a docker compose down teardown step as safety net.

Also pass context to AssetRefWriter.flush() so Redis operations
respect timeouts instead of using context.Background().

Author: GenerQAQ

.github/workflows/e2e-test.yaml

@@ -47,3 +47,9 @@ jobs:
         run: |
           cd src/server
           docker compose -f docker-compose.test.yml logs
+
+      - name: Teardown
+        if: always()
+        run: |
+          cd src/server
+          docker compose -f docker-compose.test.yml down --timeout 10

src/server/api/go/internal/infra/assetrefwriter/writer.go

@@ -112,6 +112,8 @@ func (w *AssetRefWriter) Enqueue(ctx context.Context, projectID uuid.UUID, asset
 			continue
 		}
 		pipe.SetNX(ctx, metaKey, metaJSON, defaultKeyTTL)
+		// Always refresh meta key TTL so it doesn't expire before the pending count is flushed
+		pipe.Expire(ctx, metaKey, defaultKeyTTL)
 	}
 
 	// Mark project as dirty
@@ -136,18 +138,20 @@ func (w *AssetRefWriter) loop() {
 	for {
 		select {
 		case <-ticker.C:
-			w.flush()
+			ctx, cancel := context.WithTimeout(context.Background(), w.interval)
+			w.flush(ctx)
+			cancel()
 		case <-w.stopCh:
-			// Final flush before exit
-			w.flush()
+			// Final flush before exit — use background context;
+			// Close() enforces the overall deadline via its own select.
+			w.flush(context.Background())
 			return
 		}
 	}
 }
 
 // flush pops dirty projects and flushes each one.
-func (w *AssetRefWriter) flush() {
-	ctx := context.Background()
+func (w *AssetRefWriter) flush(ctx context.Context) {
 
 	for {
 		// SPOP one project at a time to avoid holding too many in memory
@@ -222,8 +226,15 @@ func (w *AssetRefWriter) flushProject(ctx context.Context, projectID uuid.UUID)
 		metaKey := metaKeyPrefix + pid + ":" + e.sha256
 		metaJSON, err := w.redis.Get(ctx, metaKey).Bytes()
 		if err != nil {
+			if err == redis.Nil {
+				// Metadata expired or was never written — drop the orphaned pending entry
+				// to avoid infinite retry loops. The next real Enqueue will recreate it.
+				w.log.Warn("asset meta expired, dropping orphaned pending entry",
+					zap.String("key", metaKey), zap.Int64("lost_count", e.count))
+				continue
+			}
 			w.log.Error("get asset meta", zap.String("key", metaKey), zap.Error(err))
-			// Restore count to Redis since we can't flush without metadata
+			// Restore count to Redis since we can't flush without metadata (transient error)
 			w.redis.HIncrBy(ctx, pendingKey, e.sha256, e.count)
 			w.redis.SAdd(ctx, dirtySetKey, pid)
 			continue

src/server/api/go/internal/infra/assetrefwriter/writer_test.go

@@ -299,6 +299,71 @@ func TestFlush_MultipleProjects(t *testing.T) {
 	assert.Equal(t, 2, countByProject[p2])
 }
 
+func TestFlush_MissingMetadata_DropsOrphanedEntry(t *testing.T) {
+	rdb := newTestRedis(t)
+	ctx := context.Background()
+	mockRepo := &mockAssetReferenceRepo{}
+	log := zap.NewNop()
+
+	w := New(rdb, mockRepo, log, WithFlushInterval(time.Hour))
+	w.Start()
+
+	projectID := uuid.New()
+	pid := projectID.String()
+	pendingKey := pendingKeyPrefix + pid
+
+	// Manually create a pending entry WITHOUT metadata (simulates expired meta key)
+	rdb.HSet(ctx, pendingKey, "orphan_sha256", "3")
+	rdb.SAdd(ctx, dirtySetKey, pid)
+
+	// Also enqueue a valid asset
+	validAsset := model.Asset{SHA256: "valid_sha256", S3Key: "s3/valid", Bucket: "b", MIME: "application/json", SizeB: 100}
+	require.NoError(t, w.Enqueue(ctx, projectID, []model.Asset{validAsset}))
+
+	// Close triggers flush
+	w.Close(ctx)
+
+	// The valid asset should have been flushed to DB
+	require.Len(t, mockRepo.calls, 1)
+	assert.Equal(t, 1, len(mockRepo.calls[0].Assets))
+	assert.Equal(t, "valid_sha256", mockRepo.calls[0].Assets[0].SHA256)
+
+	// The orphaned entry should NOT be restored to Redis
+	count, err := rdb.HLen(ctx, pendingKey).Result()
+	require.NoError(t, err)
+	assert.Equal(t, int64(0), count, "orphaned pending entry should be dropped, not restored")
+}
+
+func TestEnqueue_RefreshesMetaTTL(t *testing.T) {
+	rdb := newTestRedis(t)
+	ctx := context.Background()
+	mockRepo := &mockAssetReferenceRepo{}
+	log := zap.NewNop()
+
+	w := New(rdb, mockRepo, log, WithFlushInterval(time.Hour))
+	w.Start()
+	defer w.Close(ctx)
+
+	projectID := uuid.New()
+	asset := model.Asset{SHA256: "aaa", S3Key: "s3/aaa", Bucket: "b"}
+
+	// First enqueue
+	require.NoError(t, w.Enqueue(ctx, projectID, []model.Asset{asset}))
+
+	metaKey := metaKeyPrefix + projectID.String() + ":aaa"
+
+	// Reduce TTL artificially to simulate time passing
+	rdb.Expire(ctx, metaKey, 1*time.Second)
+
+	// Second enqueue should refresh TTL
+	require.NoError(t, w.Enqueue(ctx, projectID, []model.Asset{asset}))
+
+	ttl, err := rdb.TTL(ctx, metaKey).Result()
+	require.NoError(t, err)
+	// TTL should have been refreshed to ~24h, not the 1s we set
+	assert.Greater(t, ttl, 1*time.Minute, "meta key TTL should be refreshed on re-enqueue")
+}
+
 func TestMetadata_SETNXIdempotent(t *testing.T) {
 	rdb := newTestRedis(t)
 	ctx := context.Background()