From b63d83a596bfa20e50a4c0dc893e68e87a8d1a24 Mon Sep 17 00:00:00 2001 From: matevip Date: Thu, 7 May 2026 16:39:39 +0800 Subject: [PATCH] feat(wiki): circuit-breaker for failing embedding provider (#72) --- .../java/vip/mate/wiki/WikiProperties.java | 15 +++++++ ...WikiEmbeddingProviderFailingException.java | 40 +++++++++++++++++++ .../wiki/service/WikiEmbeddingService.java | 33 ++++++++++++++- .../wiki/service/WikiProcessingService.java | 9 +++++ 4 files changed, 96 insertions(+), 1 deletion(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/wiki/job/WikiEmbeddingProviderFailingException.java diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java index cf3b4c00..5ec7d38e 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/WikiProperties.java @@ -130,6 +130,21 @@ public class WikiProperties { */ private int embeddingMaxChars = 6000; + /** + * Circuit-breaker threshold: abort an embedding pass after this many + * consecutive batch failures (auth / rate-limit / network errors that + * cause an entire batch to embed zero chunks). Without it, a broken + * provider would silently iterate through every pending chunk in the + * KB, producing only log noise and wasted wall-clock time before the + * user can intervene. + *

+ * Set too low and a transient blip aborts a healthy pass; set too + * high and the user waits forever on a clearly-broken provider. + * Default 5 covers most real outages while tolerating a couple of + * isolated 5xx hiccups. + */ + private int embeddingConsecutiveFailureThreshold = 5; + /** 混合搜索默认模式：keyword / semantic / hybrid */ private String searchDefaultMode = "hybrid"; diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/job/WikiEmbeddingProviderFailingException.java b/mateclaw-server/src/main/java/vip/mate/wiki/job/WikiEmbeddingProviderFailingException.java new file mode 100644 index 00000000..48964bee --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/wiki/job/WikiEmbeddingProviderFailingException.java @@ -0,0 +1,40 @@ +package vip.mate.wiki.job; + +/** + * Thrown by {@link vip.mate.wiki.service.WikiEmbeddingService#embedMissingChunks(Long)} + * when the embedding provider has failed N batches in a row, where N is + * controlled by {@code mate.wiki.embedding-consecutive-failure-threshold}. + * + *

Without this circuit, a misconfigured or unavailable provider (out + * of credits, wrong API key, network partition) silently churns through + * every pending chunk one batch at a time — producing log noise but no + * actual progress, and consuming wall-clock time the user sees as a + * stuck "task in loop". The circuit lets the embedding pass abort fast + * so the user can fix configuration and retry. + * + *

This is a soft failure: the next call into {@code embedMissingChunks} + * starts a fresh counter and will retry the provider, so once the user + * has corrected the configuration the embedding pass picks up where it + * left off without manual intervention. + */ +public class WikiEmbeddingProviderFailingException extends RuntimeException { + + private final int consecutiveFailures; + private final int remainingChunks; + + public WikiEmbeddingProviderFailingException(String message, + int consecutiveFailures, + int remainingChunks) { + super(message); + this.consecutiveFailures = consecutiveFailures; + this.remainingChunks = remainingChunks; + } + + public int getConsecutiveFailures() { + return consecutiveFailures; + } + + public int getRemainingChunks() { + return remainingChunks; + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java index 74aec2c7..b4ccf373 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiEmbeddingService.java @@ -140,7 +140,12 @@ public class WikiEmbeddingService { int batchSize = Math.max(1, properties.getEmbeddingBatchSize()); int maxChars = Math.max(500, properties.getEmbeddingMaxChars()); + int threshold = Math.max(1, properties.getEmbeddingConsecutiveFailureThreshold()); int total = 0; + // Consecutive failure counter: resets on any successful batch / long + // chunk, increments when a unit returns zero progress. Crossing the + // threshold trips the circuit and aborts the rest of this pass. + int consecutiveFailures = 0; for (int offset = 0; offset < pending.size(); offset += batchSize) { List batch = pending.subList(offset, Math.min(offset + batchSize, pending.size())); @@ -160,13 +165,30 @@ public class WikiEmbeddingService { // Short chunks: existing batch path if (!shortBatch.isEmpty()) { - total += embedShortBatch(shortBatch, r.model(), modelName, kbId); + int embedded = embedShortBatch(shortBatch, r.model(), modelName, kbId); + total += embedded; + if (embedded == 0) { + consecutiveFailures++; + if (consecutiveFailures >= threshold) { + int remaining = pending.size() - (offset + batch.size()); + throw circuitOpen(kbId, modelName, consecutiveFailures, remaining); + } + } else { + consecutiveFailures = 0; + } } // Long chunks: each goes through sub-segment split + mean pool for (WikiChunkEntity longChunk : longChunks) { if (embedLongChunk(longChunk, r.model(), modelName, maxChars)) { total++; + consecutiveFailures = 0; + } else { + consecutiveFailures++; + if (consecutiveFailures >= threshold) { + int remaining = pending.size() - (offset + batch.size()); + throw circuitOpen(kbId, modelName, consecutiveFailures, remaining); + } } } } @@ -181,6 +203,15 @@ public class WikiEmbeddingService { return total; } + private vip.mate.wiki.job.WikiEmbeddingProviderFailingException circuitOpen( + Long kbId, String modelName, int failures, int remaining) { + String message = "Embedding provider unavailable: " + failures + + " consecutive batch failures (kbId=" + kbId + ", model=" + modelName + + "). Aborted with " + remaining + " chunk(s) still pending."; + log.warn("[WikiEmbedding] Circuit opened — {}", message); + return new vip.mate.wiki.job.WikiEmbeddingProviderFailingException(message, failures, remaining); + } + /** * Embed a batch of chunks whose content fits within the per-segment char limit. * One API call per batch; individual results are persisted independently. diff --git a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java index 449f3e96..d23761f9 100644 --- a/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java +++ b/mateclaw-server/src/main/java/vip/mate/wiki/service/WikiProcessingService.java @@ -436,6 +436,12 @@ public class WikiProcessingService { if (embedded > 0) { log.info("[Wiki] Async embedding completed: kbId={}, embedded={}", fKbId, embedded); } + } catch (vip.mate.wiki.job.WikiEmbeddingProviderFailingException ex) { + // Circuit-breaker tripped — the provider has consistently failed. + // The exception's own log line in WikiEmbeddingService is enough; + // emit a calmer notice here instead of a generic failure log. + log.warn("[Wiki] Async embedding aborted by circuit-breaker for kbId={}: {}", + fKbId, ex.getMessage()); } catch (Exception ex) { log.warn("[Wiki] Async embedding failed for kbId={}: {}", fKbId, ex.getMessage()); } @@ -2306,6 +2312,9 @@ public class WikiProcessingService { if (embedded > 0) { log.info("[Wiki] Lazy async embedding completed: kbId={}, embedded={}", fKbId, embedded); } + } catch (vip.mate.wiki.job.WikiEmbeddingProviderFailingException ex) { + log.warn("[Wiki] Lazy async embedding aborted by circuit-breaker for kbId={}: {}", + fKbId, ex.getMessage()); } catch (Exception ex) { log.warn("[Wiki] Lazy async embedding failed for kbId={}: {}", fKbId, ex.getMessage()); }