新增内容，新增拉起软件层

2026-06-08 09:19:15 +08:00
parent 8ab647c419
commit c85e5b278f
67 changed files with 1552 additions and 189 deletions
@@ -83,6 +83,7 @@ public class SimilarAsinProperties {
     * 出现淘汰过频影响命中率时可上调到 512MB；2GB 堆约束下不建议超过 768MB。
     */
    private long imageCacheMaxBytes = 256L * 1024L * 1024L;
+    private boolean imageDbCacheEnabled = false;

    /**
     * 是否在 Coze 请求 parameters 中附带 api_key 字段。
@@ -12,4 +12,6 @@ public class TaskImageCacheCleanupProperties {
    private int retentionDays = 3;
    private int batchSize = 5000;
    private int maxBatchesPerRun = 200;
+    private long maxBytes = 2L * 1024L * 1024L * 1024L;
+    private long targetBytes = 1L * 1024L * 1024L * 1024L;
 }
@@ -17,4 +17,25 @@ public class TransientStorageProperties {
    private int writeTimeoutSeconds = 60;
    private int uploadMaxRetries = 3;
    private int readMaxRetries = 3;
+    private int deleteMaxRetries = 3;
+    private int maxConcurrentUploads = 16;
+    private int maxConcurrentReads = 32;
+    private int maxConcurrentDeletes = 8;
+    private long acquirePermitTimeoutMillis = 2000;
+    private long baseRetryDelayMillis = 500;
+    private long maxRetryDelayMillis = 5000;
+    private long retryJitterMillis = 250;
+    private long failureWindowSeconds = 60;
+    private int failureWindowThreshold = 20;
+    private long failureCooldownMillis = 10000;
+    private int connectionPoolMaxIdle = 0;
+    private long connectionPoolKeepAliveMillis = 1;
+    private long warnPayloadBytes = 5L * 1024 * 1024;
+    private long maxPayloadBytes = 50L * 1024 * 1024;
+    private long maxStoredPayloadBytes = 50L * 1024 * 1024;
+    private boolean fallbackToLocalOnOversize = true;
+    private boolean deleteRetryEnabled = true;
+    private String deleteRetryCron = "0 */5 * * * *";
+    private int deleteRetryQueueCapacity = 10000;
+    private int deleteRetryBatchSize = 200;
 }
@@ -33,6 +33,7 @@ public class AppearancePatentCozeClient {
    private static final String INFRINGEMENT = "侵权";
    private static final String NO_INFRINGEMENT = "无侵权";
    private static final String BRAND_QUERY_FAILED = "商标查询失败";
+    private static final String COZE_ASYNC_POLL_TIMEOUT_MESSAGE = "Coze 异步工作流轮询超时";

    private final AppearancePatentProperties properties;
    private final ObjectMapper objectMapper;
@@ -691,6 +692,18 @@ public class AppearancePatentCozeClient {
    }

    private AppearancePatentResultRowDto markFailed(AppearancePatentResultRowDto row, String failureMessage) {
+        if (isCozeAsyncPollTimeout(failureMessage)) {
+            row.setError(failureMessage);
+            if (row.getStatus() == null || row.getStatus().isBlank()) {
+                row.setStatus("FAILED");
+                row.setFailureSyntheticStatus(true);
+            }
+            row.setTitleRisk(null);
+            row.setAppearanceRisk(null);
+            row.setPatentRisk(null);
+            row.setConclusion(null);
+            return row;
+        }
        String reviewMessage = failureMessage == null || failureMessage.isBlank()
                ? "Coze 检测失败"
                : "Coze 检测失败：" + failureMessage;
@@ -718,6 +731,12 @@ public class AppearancePatentCozeClient {
        return row;
    }

+    private boolean isCozeAsyncPollTimeout(String failureMessage) {
+        String normalized = normalize(failureMessage).toLowerCase(Locale.ROOT);
+        return normalized.contains(COZE_ASYNC_POLL_TIMEOUT_MESSAGE.toLowerCase(Locale.ROOT))
+                || normalized.contains("coze async workflow poll timeout");
+    }
+
    private boolean shouldSplitBatch(List<AppearancePatentResultRowDto> rows, Exception ex) {
        return rows != null && rows.size() > 1 && isRetryableBatchFailure(ex);
    }
@@ -102,6 +102,7 @@ public class AppearancePatentTaskService {
    private static final String STATUS_RUNNING = "RUNNING";
    private static final String STATUS_SUCCESS = "SUCCESS";
    private static final String STATUS_FAILED = "FAILED";
+    private static final String COZE_ASYNC_POLL_TIMEOUT_MESSAGE = "Coze 异步工作流轮询超时";
    private static final String COZE_EMPTY_RESULT_MESSAGE = "Coze returned empty result rows";
    private static final String COZE_STATUS_SUBMITTED = "SUBMITTED";
    private static final String COZE_STATUS_RUNNING = "RUNNING";
@@ -4193,6 +4194,9 @@ public class AppearancePatentTaskService {
    }

    private String userFacingCozeCellValue(AppearancePatentResultRowDto row, String value) {
+        if (row != null && isCozeAsyncPollTimeout(row.getError())) {
+            return "";
+        }
        String normalizedValue = normalize(value);
        if (!normalizedValue.isBlank() && !isTechnicalCozeFailure(normalizedValue)) {
            return value;
@@ -4211,6 +4215,9 @@ public class AppearancePatentTaskService {
        if (row == null) {
            return "";
        }
+        if (isCozeAsyncPollTimeout(row.getError())) {
+            return "";
+        }
        String conclusion = normalize(row.getConclusion());
        if (!conclusion.isBlank() && !isTechnicalCozeFailure(conclusion)) {
            return row.getConclusion();
@@ -4225,6 +4232,12 @@ public class AppearancePatentTaskService {
        return firstNonBlank(row.getConclusion(), "");
    }

+    private boolean isCozeAsyncPollTimeout(String value) {
+        String normalized = normalize(value).toLowerCase(Locale.ROOT);
+        return normalized.contains(COZE_ASYNC_POLL_TIMEOUT_MESSAGE.toLowerCase(Locale.ROOT))
+                || normalized.contains("coze async workflow poll timeout");
+    }
+
    private String userFacingStatus(AppearancePatentResultRowDto row) {
        if (row == null) {
            return "";
@@ -0,0 +1,149 @@
+package com.nanri.aiimage.modules.file.service.object;
+
+import com.nanri.aiimage.config.TransientStorageProperties;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@Service
+@RequiredArgsConstructor
+@Slf4j
+public class RustfsDeleteRetryService {
+
+    private final TransientStorageProperties properties;
+    private final RustfsObjectStorageService rustfsObjectStorageService;
+    private final ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
+    private final ConcurrentHashMap<String, RetryItem> pending = new ConcurrentHashMap<>();
+    private final AtomicInteger queuedCount = new AtomicInteger();
+
+    public void enqueue(String objectKey, Throwable cause) {
+        if (!properties.isDeleteRetryEnabled() || objectKey == null || objectKey.isBlank()) {
+            return;
+        }
+        RetryItem item = pending.computeIfAbsent(objectKey, RetryItem::new);
+        item.lastError = cause == null ? null : cause.getMessage();
+        item.lastFailedAt = LocalDateTime.now();
+        enqueueItem(item, false);
+    }
+
+    @Scheduled(cron = "${aiimage.transient-storage.delete-retry-cron:0 */5 * * * *}")
+    public void retryPendingDeletes() {
+        if (!properties.isDeleteRetryEnabled()) {
+            return;
+        }
+        int batchSize = Math.max(1, properties.getDeleteRetryBatchSize());
+        int processed = 0;
+        int success = 0;
+        int failed = 0;
+        while (processed < batchSize) {
+            String objectKey = nextQueuedObjectKey(batchSize - processed);
+            if (objectKey == null) {
+                break;
+            }
+            RetryItem item = pending.get(objectKey);
+            if (item == null) {
+                continue;
+            }
+            item.queued = false;
+            processed++;
+            try {
+                rustfsObjectStorageService.deleteObjectFromRetry(objectKey);
+                pending.remove(objectKey);
+                success++;
+            } catch (Exception ex) {
+                failed++;
+                int failures = item.failures.incrementAndGet();
+                item.lastError = ex.getMessage();
+                item.lastFailedAt = LocalDateTime.now();
+                log.warn("[rustfs] delete retry failed objectKey={} failures={} err={}", objectKey, failures, ex.getMessage());
+                enqueueItem(item, true);
+            }
+        }
+        if (processed > 0) {
+            log.info("[rustfs] delete retry batch completed processed={} success={} failed={} pending={} queued={}",
+                    processed, success, failed, pending.size(), queuedCount.get());
+        }
+    }
+
+    int pendingCount() {
+        return pending.size();
+    }
+
+    private String nextQueuedObjectKey(int refillLimit) {
+        String objectKey = queue.poll();
+        if (objectKey != null) {
+            decrementQueuedCount();
+            refillQueue(refillLimit);
+            return objectKey;
+        }
+        if (refillQueue(refillLimit) <= 0) {
+            return null;
+        }
+        objectKey = queue.poll();
+        if (objectKey != null) {
+            decrementQueuedCount();
+            refillQueue(refillLimit);
+        }
+        return objectKey;
+    }
+
+    private int refillQueue(int maxItems) {
+        if (maxItems <= 0) {
+            return 0;
+        }
+        int capacity = Math.max(1, properties.getDeleteRetryQueueCapacity());
+        if (queuedCount.get() >= capacity) {
+            return 0;
+        }
+        int added = 0;
+        for (RetryItem item : pending.values()) {
+            if (added >= maxItems || queuedCount.get() >= capacity) {
+                break;
+            }
+            if (enqueueItem(item, true)) {
+                added++;
+            }
+        }
+        return added;
+    }
+
+    private void decrementQueuedCount() {
+        queuedCount.updateAndGet(value -> Math.max(0, value - 1));
+    }
+
+    private boolean enqueueItem(RetryItem item, boolean requeue) {
+        synchronized (item) {
+            if (item.queued) {
+                return false;
+            }
+            int capacity = Math.max(1, properties.getDeleteRetryQueueCapacity());
+            if (queuedCount.get() >= capacity) {
+                log.error("[rustfs] delete retry queue full, deferred objectKey={} requeue={} pending={} queued={} capacity={} lastError={}",
+                        item.objectKey, requeue, pending.size(), queuedCount.get(), capacity, item.lastError);
+                return false;
+            }
+            item.queued = true;
+            queue.offer(item.objectKey);
+            queuedCount.incrementAndGet();
+            return true;
+        }
+    }
+
+    private static class RetryItem {
+        private final String objectKey;
+        private final AtomicInteger failures = new AtomicInteger();
+        private volatile boolean queued;
+        private volatile String lastError;
+        private volatile LocalDateTime lastFailedAt;
+
+        private RetryItem(String objectKey) {
+            this.objectKey = objectKey;
+        }
+    }
+}
@@ -1,29 +1,71 @@
 package com.nanri.aiimage.modules.file.service.object;

 import com.nanri.aiimage.config.TransientStorageProperties;
+import io.micrometer.core.instrument.DistributionSummary;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
 import io.minio.GetObjectArgs;
 import io.minio.MinioClient;
 import io.minio.PutObjectArgs;
 import io.minio.RemoveObjectArgs;
 import io.minio.StatObjectArgs;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import okhttp3.ConnectionPool;
 import okhttp3.OkHttpClient;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;

 import java.io.ByteArrayInputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;

@Service
-@RequiredArgsConstructor
@Slf4j
 public class RustfsObjectStorageService {

+    private static final String OP_UPLOAD = "upload";
+    private static final String OP_READ = "read";
+    private static final String OP_DELETE = "delete";
+    private static final String OP_STAT = "stat";
+
    private final TransientStorageProperties properties;
+    private final ObjectProvider<MeterRegistry> meterRegistryProvider;
+    private final ObjectProvider<RustfsDeleteRetryService> deleteRetryServiceProvider;
+    private final Supplier<MinioClient> minioClientSupplier;
+    private final Semaphore uploadSemaphore;
+    private final Semaphore readSemaphore;
+    private final Semaphore deleteSemaphore;
+    private final AtomicInteger windowFailureCount = new AtomicInteger();
+    private volatile long failureWindowStartedAtMillis;
+    private volatile long circuitOpenUntilMillis;
    private volatile OkHttpClient httpClient;

+    @Autowired
+    public RustfsObjectStorageService(TransientStorageProperties properties,
+                                      ObjectProvider<MeterRegistry> meterRegistryProvider,
+                                      ObjectProvider<RustfsDeleteRetryService> deleteRetryServiceProvider) {
+        this(properties, meterRegistryProvider, deleteRetryServiceProvider, null);
+    }
+
+    RustfsObjectStorageService(TransientStorageProperties properties,
+                               ObjectProvider<MeterRegistry> meterRegistryProvider,
+                               ObjectProvider<RustfsDeleteRetryService> deleteRetryServiceProvider,
+                               Supplier<MinioClient> minioClientSupplier) {
+        this.properties = properties;
+        this.meterRegistryProvider = meterRegistryProvider;
+        this.deleteRetryServiceProvider = deleteRetryServiceProvider;
+        this.minioClientSupplier = minioClientSupplier;
+        this.uploadSemaphore = new Semaphore(Math.max(1, properties.getMaxConcurrentUploads()));
+        this.readSemaphore = new Semaphore(Math.max(1, properties.getMaxConcurrentReads()));
+        this.deleteSemaphore = new Semaphore(Math.max(1, properties.getMaxConcurrentDeletes()));
+    }
+
    public boolean isConfigured() {
        return notBlank(properties.getEndpoint())
                && notBlank(properties.getBucket())
@@ -39,10 +81,10 @@ public class RustfsObjectStorageService {
        if (!isConfigured()) {
            throw new IllegalStateException("transient storage is not configured");
        }
+        rejectIfCircuitOpen(OP_UPLOAD, objectKey);
        byte[] bytes = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8);
-        Exception last = null;
-        int maxRetries = Math.max(1, properties.getUploadMaxRetries());
-        for (int attempt = 1; attempt <= maxRetries; attempt++) {
+        recordPayloadBytes(bytes.length);
+        return executeWithRetry(OP_UPLOAD, objectKey, bytes.length, Math.max(1, properties.getUploadMaxRetries()), uploadSemaphore, () -> {
            try (ByteArrayInputStream stream = new ByteArrayInputStream(bytes)) {
                buildClient().putObject(PutObjectArgs.builder()
                        .bucket(properties.getBucket())
@@ -54,56 +96,117 @@ public class RustfsObjectStorageService {
                    verifyObjectVisible(objectKey);
                }
                return objectKey;
-            } catch (Exception ex) {
-                last = ex;
-                if (attempt < maxRetries) {
-                    log.warn("[rustfs] upload failed, retrying objectKey={} attempt={}/{}", objectKey, attempt, maxRetries, ex);
-                    sleepQuietly(500L * attempt);
-                }
            }
-        }
-        throw new IllegalStateException("failed to upload payload to transient storage", last);
+        });
    }

    public String readObjectAsString(String objectKey) {
        if (!isConfigured()) {
            throw new IllegalStateException("transient storage is not configured");
        }
-        Exception last = null;
-        int maxRetries = Math.max(1, properties.getReadMaxRetries());
-        for (int attempt = 1; attempt <= maxRetries; attempt++) {
+        rejectIfCircuitOpen(OP_READ, objectKey);
+        return executeWithRetry(OP_READ, objectKey, 0, Math.max(1, properties.getReadMaxRetries()), readSemaphore, () -> {
            try (var stream = buildClient().getObject(GetObjectArgs.builder()
                    .bucket(properties.getBucket())
                    .object(objectKey)
                    .build())) {
                return new String(stream.readAllBytes(), StandardCharsets.UTF_8);
-            } catch (Exception ex) {
-                last = ex;
-                if (attempt < maxRetries) {
-                    // 与 uploadText 对齐的线性退避，避免 rustfs 短暂抖动直接打成 read coze batch failed。
-                    log.warn("[rustfs] read failed, retrying objectKey={} attempt={}/{} err={}", objectKey, attempt, maxRetries, ex.getMessage());
-                    sleepQuietly(500L * attempt);
-                }
            }
-        }
-        throw new IllegalStateException("failed to read payload from transient storage", last);
+        });
    }

    public void deleteObject(String objectKey) {
+        deleteObject(objectKey, true);
+    }
+
+    void deleteObjectFromRetry(String objectKey) {
+        deleteObject(objectKey, false);
+    }
+
+    private void deleteObject(String objectKey, boolean enqueueRetryOnFailure) {
        if (!isConfigured() || !notBlank(objectKey)) {
            return;
        }
        try {
-            buildClient().removeObject(RemoveObjectArgs.builder()
-                    .bucket(properties.getBucket())
-                    .object(objectKey)
-                    .build());
-        } catch (Exception ex) {
-            throw new IllegalStateException("failed to delete payload from transient storage", ex);
+            executeWithRetry(OP_DELETE, objectKey, 0, Math.max(1, properties.getDeleteMaxRetries()), deleteSemaphore, () -> {
+                buildClient().removeObject(RemoveObjectArgs.builder()
+                        .bucket(properties.getBucket())
+                        .object(objectKey)
+                        .build());
+                return null;
+            });
+        } catch (RuntimeException ex) {
+            if (enqueueRetryOnFailure && properties.isDeleteRetryEnabled()) {
+                RustfsDeleteRetryService retryService = deleteRetryServiceProvider == null ? null : deleteRetryServiceProvider.getIfAvailable();
+                if (retryService != null) {
+                    retryService.enqueue(objectKey, ex);
+                }
+            }
+            throw ex;
+        }
+    }
+
+    private <T> T executeWithRetry(String operation,
+                                   String objectKey,
+                                   long bytes,
+                                   int maxRetries,
+                                   Semaphore semaphore,
+                                   CheckedSupplier<T> supplier) {
+        acquirePermit(operation, objectKey, semaphore);
+        long startedAt = System.nanoTime();
+        Exception last = null;
+        try {
+            for (int attempt = 1; attempt <= maxRetries; attempt++) {
+                try {
+                    T result = supplier.get();
+                    resetFailureWindow();
+                    recordOperation(operation, "success", elapsedNanos(startedAt));
+                    log.debug("[rustfs] operation success operation={} objectKey={} attempt={}/{} durationMs={} bytes={}",
+                            operation, objectKey, attempt, maxRetries, elapsedMillis(startedAt), bytes);
+                    return result;
+                } catch (Exception ex) {
+                    last = ex;
+                    evictIdleConnections();
+                    recordOperation(operation, attempt < maxRetries ? "retry" : "failure", elapsedNanos(startedAt));
+                    recordFailure(operation, objectKey, ex);
+                    if (attempt < maxRetries) {
+                        long delayMillis = retryDelayMillis(attempt);
+                        log.warn("[rustfs] operation failed, retrying operation={} objectKey={} attempt={}/{} delayMs={} err={}",
+                                operation, objectKey, attempt, maxRetries, delayMillis, ex.getMessage());
+                        sleepQuietly(delayMillis, "retrying rustfs " + operation);
+                    }
+                }
+            }
+        } finally {
+            semaphore.release();
+        }
+        throw new IllegalStateException("failed to " + operation + " payload in transient storage", last);
+    }
+
+    private void acquirePermit(String operation, String objectKey, Semaphore semaphore) {
+        try {
+            boolean acquired;
+            long timeoutMillis = Math.max(0L, properties.getAcquirePermitTimeoutMillis());
+            if (timeoutMillis == 0L) {
+                acquired = semaphore.tryAcquire();
+            } else {
+                acquired = semaphore.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
+            }
+            if (!acquired) {
+                recordOperation(operation, "rejected", 0L);
+                throw new IllegalStateException("rustfs " + operation + " concurrency limit reached: " + objectKey);
+            }
+        } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
+            recordOperation(operation, "rejected", 0L);
+            throw new IllegalStateException("interrupted while acquiring rustfs " + operation + " permit", ex);
        }
    }

    private MinioClient buildClient() {
+        if (minioClientSupplier != null) {
+            return minioClientSupplier.get();
+        }
        return MinioClient.builder()
                .endpoint(properties.getEndpoint())
                .credentials(properties.getAccessKeyId(), properties.getAccessKeySecret())
@@ -112,7 +215,7 @@ public class RustfsObjectStorageService {
                .build();
    }

-    private OkHttpClient getHttpClient() {
+    OkHttpClient getHttpClient() {
        OkHttpClient current = httpClient;
        if (current != null) {
            return current;
@@ -123,13 +226,26 @@ public class RustfsObjectStorageService {
                        .connectTimeout(Math.max(1, properties.getConnectTimeoutSeconds()), TimeUnit.SECONDS)
                        .readTimeout(Math.max(1, properties.getReadTimeoutSeconds()), TimeUnit.SECONDS)
                        .writeTimeout(Math.max(1, properties.getWriteTimeoutSeconds()), TimeUnit.SECONDS)
+                        .connectionPool(new ConnectionPool(
+                                Math.max(0, properties.getConnectionPoolMaxIdle()),
+                                Math.max(1L, properties.getConnectionPoolKeepAliveMillis()),
+                                TimeUnit.MILLISECONDS))
+                        .retryOnConnectionFailure(true)
                        .build();
            }
            return httpClient;
        }
    }

+    private void evictIdleConnections() {
+        OkHttpClient current = httpClient;
+        if (current != null) {
+            current.connectionPool().evictAll();
+        }
+    }
+
    private void verifyObjectVisible(String objectKey) {
+        long startedAt = System.nanoTime();
        RuntimeException last = null;
        for (int attempt = 1; attempt <= 3; attempt++) {
            try {
@@ -137,11 +253,14 @@ public class RustfsObjectStorageService {
                        .bucket(properties.getBucket())
                        .object(objectKey)
                        .build());
+                recordOperation(OP_STAT, "success", elapsedNanos(startedAt));
                return;
            } catch (Exception ex) {
                last = new IllegalStateException("transient payload is not visible after upload: " + objectKey, ex);
+                evictIdleConnections();
+                recordOperation(OP_STAT, attempt < 3 ? "retry" : "failure", elapsedNanos(startedAt));
                if (attempt < 3) {
-                    sleepQuietly(100L * attempt);
+                    sleepQuietly(100L * attempt, "verifying transient payload upload");
                }
            }
        }
@@ -150,16 +269,117 @@ public class RustfsObjectStorageService {
                : last;
    }

-    private void sleepQuietly(long millis) {
+    private void rejectIfCircuitOpen(String operation, String objectKey) {
+        long openUntil = circuitOpenUntilMillis;
+        if (openUntil <= 0L) {
+            return;
+        }
+        long now = System.currentTimeMillis();
+        if (now < openUntil) {
+            recordOperation(operation, "rejected", 0L);
+            throw new IllegalStateException("rustfs failure cooldown active operation=" + operation + " objectKey=" + objectKey);
+        }
+    }
+
+    private void recordFailure(String operation, String objectKey, Exception ex) {
+        long now = System.currentTimeMillis();
+        long windowMillis = Math.max(1L, properties.getFailureWindowSeconds()) * 1000L;
+        if (failureWindowStartedAtMillis <= 0L || now - failureWindowStartedAtMillis > windowMillis) {
+            failureWindowStartedAtMillis = now;
+            windowFailureCount.set(0);
+        }
+        int failures = windowFailureCount.incrementAndGet();
+        int threshold = Math.max(1, properties.getFailureWindowThreshold());
+        if ((OP_UPLOAD.equals(operation) || OP_READ.equals(operation)) && failures >= threshold) {
+            long cooldownMillis = Math.max(0L, properties.getFailureCooldownMillis());
+            circuitOpenUntilMillis = now + cooldownMillis;
+            windowFailureCount.set(0);
+            failureWindowStartedAtMillis = now;
+            log.error("[rustfs] failure window threshold reached, cooldown enabled operation={} objectKey={} failures={} threshold={} cooldownMs={} err={}",
+                    operation, objectKey, failures, threshold, cooldownMillis, ex.getMessage());
+        }
+    }
+
+    private void resetFailureWindow() {
+        windowFailureCount.set(0);
+        failureWindowStartedAtMillis = 0L;
+        circuitOpenUntilMillis = 0L;
+    }
+
+    private long retryDelayMillis(int attempt) {
+        long baseDelay = Math.max(0L, properties.getBaseRetryDelayMillis());
+        long maxDelay = Math.max(baseDelay, properties.getMaxRetryDelayMillis());
+        long delay = Math.min(maxDelay, baseDelay * Math.max(1, attempt));
+        long jitter = Math.max(0L, properties.getRetryJitterMillis());
+        if (jitter > 0L) {
+            delay += ThreadLocalRandom.current().nextLong(jitter + 1L);
+        }
+        return delay;
+    }
+
+    private void sleepQuietly(long millis, String reason) {
+        if (millis <= 0L) {
+            return;
+        }
        try {
            Thread.sleep(millis);
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
-            throw new IllegalStateException("interrupted while verifying transient payload upload", ex);
+            throw new IllegalStateException("interrupted while " + reason, ex);
        }
    }

+    private void recordPayloadBytes(long bytes) {
+        MeterRegistry registry = meterRegistry();
+        if (registry == null || bytes < 0L) {
+            return;
+        }
+        DistributionSummary.builder("aiimage.rustfs.payload.bytes")
+                .baseUnit("bytes")
+                .register(registry)
+                .record(bytes);
+    }
+
+    private void recordOperation(String operation, String result, long durationNanos) {
+        MeterRegistry registry = meterRegistry();
+        if (registry == null) {
+            return;
+        }
+        registry.counter("aiimage.rustfs.operation.total", "operation", operation, "result", result).increment();
+        if (durationNanos > 0L) {
+            Timer.builder("aiimage.rustfs.operation.duration")
+                    .tag("operation", operation)
+                    .tag("result", result)
+                    .register(registry)
+                    .record(durationNanos, TimeUnit.NANOSECONDS);
+        }
+    }
+
+    public void recordLocalFallback() {
+        MeterRegistry registry = meterRegistry();
+        if (registry != null) {
+            registry.counter("aiimage.rustfs.fallback.local.total").increment();
+        }
+    }
+
+    private MeterRegistry meterRegistry() {
+        return meterRegistryProvider == null ? null : meterRegistryProvider.getIfAvailable();
+    }
+
+    private long elapsedNanos(long startedAt) {
+        return System.nanoTime() - startedAt;
+    }
+
+    private long elapsedMillis(long startedAt) {
+        return TimeUnit.NANOSECONDS.toMillis(elapsedNanos(startedAt));
+    }
+
    private boolean notBlank(String value) {
        return value != null && !value.isBlank();
    }
+
+    @FunctionalInterface
+    private interface CheckedSupplier<T> {
+        T get() throws Exception;
+    }
 }
@@ -1200,5 +1200,3 @@ public class QueryAsinTaskService {
        }
    }
 }
-
-
@@ -1,6 +1,7 @@
 package com.nanri.aiimage.modules.similarasin.service;

 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.nanri.aiimage.config.SimilarAsinProperties;
 import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder;
 import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder.ResizedImage;
 import com.nanri.aiimage.modules.task.mapper.TaskImageCacheMapper;
@@ -54,6 +55,7 @@ public class SimilarAsinImagePrefetchService {

    private final SimilarAsinImageEmbedder imageEmbedder;
    private final TaskImageCacheMapper taskImageCacheMapper;
+    private final SimilarAsinProperties properties;

    /**
     * 每个 task 当前 in-flight 的预热 future。enqueue 时如果上一个还没完成，会先等它结束，
@@ -75,6 +77,9 @@ public class SimilarAsinImagePrefetchService {
     * 同 task 串行入队（用 inflight map 排队），避免多个 batch 同时打爆图片源站。
     */
    public void enqueue(Long taskId, List<String> urls) {
+        if (!properties.isImageDbCacheEnabled()) {
+            return;
+        }
        if (taskId == null || urls == null || urls.isEmpty()) {
            return;
        }
@@ -165,6 +170,9 @@ public class SimilarAsinImagePrefetchService {
     * 失败/未命中返回 null，由调用方走回退路径。
     */
    public byte[] lookup(String url) {
+        if (!properties.isImageDbCacheEnabled()) {
+            return null;
+        }
        if (url == null) {
            return null;
        }
@@ -30,6 +30,15 @@ public interface TaskImageCacheMapper extends BaseMapper<TaskImageCacheEntity> {
    @Select("SELECT image_bytes FROM biz_task_image_cache WHERE url_hash = #{urlHash} LIMIT 1")
    byte[] selectBytesByUrlHash(@Param("urlHash") String urlHash);

+    @Select("SELECT COALESCE(SUM(byte_size), 0) FROM biz_task_image_cache")
+    Long sumByteSize();
+
+    @Select("SELECT COALESCE(SUM(byte_size), 0) FROM (SELECT byte_size FROM biz_task_image_cache ORDER BY last_used_at LIMIT #{limit}) oldest")
+    Long sumOldestBatchByteSize(@Param("limit") int limit);
+
    @Delete("DELETE FROM biz_task_image_cache WHERE last_used_at < #{cutoff} ORDER BY last_used_at LIMIT #{limit}")
    int deleteExpiredBatch(@Param("cutoff") LocalDateTime cutoff, @Param("limit") int limit);
+
+    @Delete("DELETE FROM biz_task_image_cache ORDER BY last_used_at LIMIT #{limit}")
+    int deleteOldestBatch(@Param("limit") int limit);
 }
@@ -39,18 +39,18 @@ public class TaskImageCacheCleanupService {
            int maxBatches = Math.max(1, cleanupProperties.getMaxBatchesPerRun());
            LocalDateTime cutoff = LocalDateTime.now().minusDays(retentionDays);

-            int deletedTotal = 0;
-            int batches = 0;
-            while (batches < maxBatches) {
+            int expiredDeletedTotal = 0;
+            int expiredBatches = 0;
+            while (expiredBatches < maxBatches) {
                int deleted = taskImageCacheMapper.deleteExpiredBatch(cutoff, batchSize);
                if (deleted <= 0) {
                    break;
                }
-                deletedTotal += deleted;
-                batches++;
+                expiredDeletedTotal += deleted;
+                expiredBatches++;
                if (!lockHandle.renew(CLEANUP_LOCK_TTL)) {
                    log.warn("[task-image-cache-cleanup] stop because lock renew failed deletedTotal={} batches={}",
-                            deletedTotal, batches);
+                            expiredDeletedTotal, expiredBatches);
                    break;
                }
                if (deleted < batchSize) {
@@ -58,12 +58,49 @@ public class TaskImageCacheCleanupService {
                }
            }

-            if (deletedTotal > 0 || batches >= maxBatches) {
-                log.info("[task-image-cache-cleanup] finished cutoff={} retentionDays={} deletedTotal={} batches={} batchSize={} maxBatches={}",
-                        cutoff, retentionDays, deletedTotal, batches, batchSize, maxBatches);
+            long maxBytes = cleanupProperties.getMaxBytes();
+            long targetBytes = cleanupProperties.getTargetBytes();
+            int capacityDeletedTotal = 0;
+            int capacityBatches = 0;
+            if (maxBytes > 0L) {
+                long actualTargetBytes = targetBytes > 0L && targetBytes < maxBytes ? targetBytes : maxBytes / 2L;
+                long currentBytes = nullToZero(taskImageCacheMapper.sumByteSize());
+                boolean overCapacity = currentBytes > maxBytes;
+                while (overCapacity && currentBytes > actualTargetBytes && capacityBatches < maxBatches) {
+                    long batchBytes = nullToZero(taskImageCacheMapper.sumOldestBatchByteSize(batchSize));
+                    if (batchBytes <= 0L) {
+                        break;
+                    }
+                    int deleted = taskImageCacheMapper.deleteOldestBatch(batchSize);
+                    if (deleted <= 0) {
+                        break;
+                    }
+                    capacityDeletedTotal += deleted;
+                    capacityBatches++;
+                    currentBytes = Math.max(0L, currentBytes - batchBytes);
+                    if (!lockHandle.renew(CLEANUP_LOCK_TTL)) {
+                        log.warn("[task-image-cache-cleanup] stop capacity cleanup because lock renew failed deletedTotal={} batches={}",
+                                capacityDeletedTotal, capacityBatches);
+                        break;
+                    }
+                    if (deleted < batchSize) {
+                        break;
+                    }
+                }
+            }
+
+            if (expiredDeletedTotal > 0 || expiredBatches >= maxBatches
+                    || capacityDeletedTotal > 0 || capacityBatches >= maxBatches) {
+                log.info("[task-image-cache-cleanup] finished cutoff={} retentionDays={} expiredDeleted={} expiredBatches={} capacityDeleted={} capacityBatches={} batchSize={} maxBatches={} maxBytes={} targetBytes={}",
+                        cutoff, retentionDays, expiredDeletedTotal, expiredBatches, capacityDeletedTotal,
+                        capacityBatches, batchSize, maxBatches, maxBytes, targetBytes);
            }
        } catch (Exception ex) {
            log.warn("[task-image-cache-cleanup] failed msg={}", ex.getMessage(), ex);
        }
    }
+
+    private static long nullToZero(Long value) {
+        return value == null ? 0L : value;
+    }
 }
@@ -312,20 +312,39 @@ public class TransientPayloadStorageService {
            return content;
        }
        String objectKey = buildObjectKey(category, moduleType, taskId, scopeHash, entryKey);
+        long rawBytes = payloadBytes(content);
+        if (isPositiveLimit(properties.getWarnPayloadBytes()) && rawBytes > properties.getWarnPayloadBytes()) {
+            log.warn("[transient-payload] payload size exceeds warn threshold category={} moduleType={} taskId={} objectKey={} rawBytes={} warnBytes={}",
+                    category, moduleType, taskId, objectKey, rawBytes, properties.getWarnPayloadBytes());
+        }
+        boolean rawOversize = isPositiveLimit(properties.getMaxPayloadBytes()) && rawBytes > properties.getMaxPayloadBytes();
        String storedContent = encodeStoredPayload(content);
+        long storedBytes = payloadBytes(storedContent);
+        boolean storedOversize = isPositiveLimit(properties.getMaxStoredPayloadBytes()) && storedBytes > properties.getMaxStoredPayloadBytes();
+        if (rawOversize || storedOversize) {
+            log.warn("[transient-payload] payload size exceeds rustfs limit category={} moduleType={} taskId={} objectKey={} rawBytes={} storedBytes={} maxRawBytes={} maxStoredBytes={} fallbackToLocal={}",
+                    category, moduleType, taskId, objectKey, rawBytes, storedBytes,
+                    properties.getMaxPayloadBytes(), properties.getMaxStoredPayloadBytes(), properties.isFallbackToLocalOnOversize());
+            if (!properties.isFallbackToLocalOnOversize()) {
+                throw new IllegalStateException("transient payload exceeds configured size limit: " + objectKey);
+            }
+        }
        String pointer = null;
-        boolean rustfsFallbackToLocal = false;
-        if (rustfsObjectStorageService.isConfigured()) {
+        boolean rustfsFallbackToLocal = rawOversize || storedOversize;
+        if (!rustfsFallbackToLocal && rustfsObjectStorageService.isConfigured()) {
            try {
                pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, storedContent, verifyAfterUpload);
            } catch (Exception ex) {
                rustfsFallbackToLocal = true;
                // 升级为 ERROR：rustfs 失败后只能落到本地，多实例下其他节点读不到，必须能告警。
-                log.error("[transient-payload] rustfs upload failed, fallback to local store instanceId={} category={} taskId={} objectKey={} err={}",
-                        instanceMetadata.getInstanceId(), category, taskId, objectKey, ex.getMessage());
+                log.error("[transient-payload] rustfs upload failed, fallback to local store instanceId={} category={} moduleType={} taskId={} objectKey={} rawBytes={} storedBytes={} err={}",
+                        instanceMetadata.getInstanceId(), category, moduleType, taskId, objectKey, rawBytes, storedBytes, ex.getMessage());
            }
        }
        if (pointer == null) {
+            if (rustfsFallbackToLocal) {
+                rustfsObjectStorageService.recordLocalFallback();
+            }
            pointer = storeLocal(objectKey, storedContent);
        }
        // P0-3：记录本次 store 是否走 local 兜底，供调用方在拿到 pointer 后立即查询。
@@ -429,6 +448,14 @@ public class TransientPayloadStorageService {
        }
    }

+    private long payloadBytes(String content) {
+        return Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8).length;
+    }
+
+    private boolean isPositiveLimit(long value) {
+        return value > 0L;
+    }
+
    private String decodeStoredPayload(String storedContent) {
        if (storedContent == null || storedContent.isBlank()) {
            return storedContent;