Merge branch 'fix/approve-serializable-retry' into 'main'
fix(approve): retry на Postgres SERIALIZABLE 40001 (массовый approve drafts) See merge request 2-6/2-6-4/terravault/ordinis!286
This commit is contained in:
+83
@@ -0,0 +1,83 @@
|
|||||||
|
package cloud.nstart.terravault.ordinis.restapi.service;
|
||||||
|
|
||||||
|
import java.util.concurrent.ThreadLocalRandom;
|
||||||
|
import java.util.function.Supplier;
|
||||||
|
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.dao.CannotAcquireLockException;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Retry helper для SERIALIZABLE-isolation транзакций (DraftService.approve /
|
||||||
|
* reject и SchemaDraftService.publish).
|
||||||
|
*
|
||||||
|
* <p>Postgres при isolation=SERIALIZABLE использует SSI (Serializable
|
||||||
|
* Snapshot Isolation). Когда две транзакции читают/пишут пересекающиеся
|
||||||
|
* read-write зависимости, одна из них кэнселится с {@code SQLSTATE 40001}:
|
||||||
|
*
|
||||||
|
* <pre>{@code
|
||||||
|
* ERROR: could not serialize access due to read/write dependencies among transactions
|
||||||
|
* Hint: The transaction might succeed if retried.
|
||||||
|
* }</pre>
|
||||||
|
*
|
||||||
|
* <p>Это <b>transient</b> ошибка — Postgres сам в Hint говорит retry. Без
|
||||||
|
* retry даже корректный код упадёт 500 при concurrent approve drafts (юзер
|
||||||
|
* выбрал 10 drafts → bulk approve fan-out → 1-2 транзакции aborted).
|
||||||
|
*
|
||||||
|
* <p><b>ВАЖНО:</b> этот helper должен вызываться <b>ИЗ ВНЕ</b> @Transactional
|
||||||
|
* boundary (например, из Controller'а). Если retry внутри @Transactional —
|
||||||
|
* каждый attempt будет запускаться в той же aborted транзакции и снова
|
||||||
|
* упадёт. Service'ы остаются @Transactional как раньше — controller
|
||||||
|
* заворачивает их вызов.
|
||||||
|
*
|
||||||
|
* <p>Backoff: exponential (50ms / 100ms / 200ms) + jitter [0..50ms). После
|
||||||
|
* 3 неуспешных попыток — пробрасываем последнюю CannotAcquireLockException
|
||||||
|
* → GlobalExceptionHandler → 500. Это очень редкий case (3 retries при
|
||||||
|
* high contention) — лучше fail чем infinite loop.
|
||||||
|
*/
|
||||||
|
public final class SerializableRetry {
|
||||||
|
private static final Logger log = LoggerFactory.getLogger(SerializableRetry.class);
|
||||||
|
private static final int MAX_ATTEMPTS = 3;
|
||||||
|
private static final long BASE_DELAY_MS = 50;
|
||||||
|
|
||||||
|
private SerializableRetry() {}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Запустить операцию с retry на {@link CannotAcquireLockException} (Spring
|
||||||
|
* translation для Postgres 40001 serialization_failure).
|
||||||
|
*
|
||||||
|
* @param op операция — обычно лямбда вызывающая service.approve(...)
|
||||||
|
* @return результат успешной попытки
|
||||||
|
* @throws CannotAcquireLockException если все {@value #MAX_ATTEMPTS}
|
||||||
|
* попыток упали (бросаем последнюю — GlobalExceptionHandler
|
||||||
|
* маппит в 500)
|
||||||
|
*/
|
||||||
|
public static <T> T execute(Supplier<T> op) {
|
||||||
|
CannotAcquireLockException last = null;
|
||||||
|
for (int attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
|
||||||
|
try {
|
||||||
|
return op.get();
|
||||||
|
} catch (CannotAcquireLockException e) {
|
||||||
|
last = e;
|
||||||
|
if (attempt < MAX_ATTEMPTS - 1) {
|
||||||
|
long backoff = (1L << attempt) * BASE_DELAY_MS;
|
||||||
|
long jitter = ThreadLocalRandom.current().nextLong(0, BASE_DELAY_MS);
|
||||||
|
long sleepMs = backoff + jitter;
|
||||||
|
log.warn(
|
||||||
|
"Postgres SERIALIZABLE conflict; retry {}/{} after {}ms",
|
||||||
|
attempt + 1, MAX_ATTEMPTS, sleepMs);
|
||||||
|
try {
|
||||||
|
Thread.sleep(sleepMs);
|
||||||
|
} catch (InterruptedException ie) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
throw e;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
log.error(
|
||||||
|
"Postgres SERIALIZABLE conflict: exhausted {} attempts; surfacing 500",
|
||||||
|
MAX_ATTEMPTS, last);
|
||||||
|
throw last;
|
||||||
|
}
|
||||||
|
}
|
||||||
+9
-2
@@ -2,6 +2,7 @@ package cloud.nstart.terravault.ordinis.restapi.web;
|
|||||||
|
|
||||||
import cloud.nstart.terravault.ordinis.restapi.dto.draft.DraftResponse;
|
import cloud.nstart.terravault.ordinis.restapi.dto.draft.DraftResponse;
|
||||||
import cloud.nstart.terravault.ordinis.restapi.dto.draft.SubmitDraftRequest;
|
import cloud.nstart.terravault.ordinis.restapi.dto.draft.SubmitDraftRequest;
|
||||||
|
import cloud.nstart.terravault.ordinis.restapi.service.SerializableRetry;
|
||||||
import cloud.nstart.terravault.ordinis.restapi.service.draft.DraftService;
|
import cloud.nstart.terravault.ordinis.restapi.service.draft.DraftService;
|
||||||
import jakarta.validation.Valid;
|
import jakarta.validation.Valid;
|
||||||
import org.springframework.data.domain.PageRequest;
|
import org.springframework.data.domain.PageRequest;
|
||||||
@@ -89,7 +90,12 @@ public class DraftController {
|
|||||||
public DraftResponse approve(
|
public DraftResponse approve(
|
||||||
@PathVariable UUID id,
|
@PathVariable UUID id,
|
||||||
@RequestParam(required = false) String comment) {
|
@RequestParam(required = false) String comment) {
|
||||||
return DraftResponse.from(draftService.approve(id, comment));
|
// SERIALIZABLE retry — при bulk approve N drafts (UI fan-out)
|
||||||
|
// Postgres SSI кэнселит concurrent transactions с 40001. Без retry
|
||||||
|
// юзер получит 500 на random subset. Retry — outside @Transactional
|
||||||
|
// (каждый attempt — fresh tx).
|
||||||
|
return DraftResponse.from(
|
||||||
|
SerializableRetry.execute(() -> draftService.approve(id, comment)));
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -100,7 +106,8 @@ public class DraftController {
|
|||||||
public DraftResponse reject(
|
public DraftResponse reject(
|
||||||
@PathVariable UUID id,
|
@PathVariable UUID id,
|
||||||
@RequestParam String comment) {
|
@RequestParam String comment) {
|
||||||
return DraftResponse.from(draftService.reject(id, comment));
|
return DraftResponse.from(
|
||||||
|
SerializableRetry.execute(() -> draftService.reject(id, comment)));
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
+5
-2
@@ -9,6 +9,7 @@ import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.SchemaDraftRes
|
|||||||
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.SubmitForReviewRequest;
|
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.SubmitForReviewRequest;
|
||||||
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.UpdateSchemaDraftRequest;
|
import cloud.nstart.terravault.ordinis.restapi.dto.schemaworkflow.UpdateSchemaDraftRequest;
|
||||||
import cloud.nstart.terravault.ordinis.restapi.service.DictionaryDefinitionService;
|
import cloud.nstart.terravault.ordinis.restapi.service.DictionaryDefinitionService;
|
||||||
|
import cloud.nstart.terravault.ordinis.restapi.service.SerializableRetry;
|
||||||
import cloud.nstart.terravault.ordinis.restapi.service.schemaworkflow.SchemaDraftService;
|
import cloud.nstart.terravault.ordinis.restapi.service.schemaworkflow.SchemaDraftService;
|
||||||
import jakarta.validation.Valid;
|
import jakarta.validation.Valid;
|
||||||
import org.springframework.data.domain.PageRequest;
|
import org.springframework.data.domain.PageRequest;
|
||||||
@@ -107,7 +108,8 @@ public class SchemaDraftController {
|
|||||||
@PathVariable String name,
|
@PathVariable String name,
|
||||||
@PathVariable UUID draftId,
|
@PathVariable UUID draftId,
|
||||||
@Valid @RequestBody DecisionRequest req) {
|
@Valid @RequestBody DecisionRequest req) {
|
||||||
DictionarySchemaDraft draft = draftService.decide(draftId, req);
|
DictionarySchemaDraft draft = SerializableRetry.execute(
|
||||||
|
() -> draftService.decide(draftId, req));
|
||||||
return SchemaDraftResponse.from(draft, name);
|
return SchemaDraftResponse.from(draft, name);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -125,7 +127,8 @@ public class SchemaDraftController {
|
|||||||
@PathVariable String name,
|
@PathVariable String name,
|
||||||
@PathVariable UUID draftId,
|
@PathVariable UUID draftId,
|
||||||
@Valid @RequestBody PublishDraftRequest req) {
|
@Valid @RequestBody PublishDraftRequest req) {
|
||||||
DictionarySchemaDraft draft = draftService.publish(draftId, req);
|
DictionarySchemaDraft draft = SerializableRetry.execute(
|
||||||
|
() -> draftService.publish(draftId, req));
|
||||||
return Map.of(
|
return Map.of(
|
||||||
"draft", SchemaDraftResponse.from(draft, name),
|
"draft", SchemaDraftResponse.from(draft, name),
|
||||||
"newVersion", draft.getPublishedVersion(),
|
"newVersion", draft.getPublishedVersion(),
|
||||||
|
|||||||
+57
@@ -0,0 +1,57 @@
|
|||||||
|
package cloud.nstart.terravault.ordinis.restapi.service;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||||
|
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.dao.CannotAcquireLockException;
|
||||||
|
|
||||||
|
class SerializableRetryTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void returnsResultOnFirstAttemptWhenNoConflict() {
|
||||||
|
AtomicInteger calls = new AtomicInteger();
|
||||||
|
String result = SerializableRetry.execute(() -> {
|
||||||
|
calls.incrementAndGet();
|
||||||
|
return "ok";
|
||||||
|
});
|
||||||
|
assertThat(result).isEqualTo("ok");
|
||||||
|
assertThat(calls.get()).isEqualTo(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void retriesOnSerializableConflictAndSucceeds() {
|
||||||
|
AtomicInteger calls = new AtomicInteger();
|
||||||
|
String result = SerializableRetry.execute(() -> {
|
||||||
|
int attempt = calls.incrementAndGet();
|
||||||
|
if (attempt < 3) {
|
||||||
|
throw new CannotAcquireLockException("40001 serialization_failure");
|
||||||
|
}
|
||||||
|
return "ok";
|
||||||
|
});
|
||||||
|
assertThat(result).isEqualTo("ok");
|
||||||
|
assertThat(calls.get()).isEqualTo(3);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void surfacesExceptionAfterMaxAttempts() {
|
||||||
|
AtomicInteger calls = new AtomicInteger();
|
||||||
|
assertThatThrownBy(() -> SerializableRetry.execute(() -> {
|
||||||
|
calls.incrementAndGet();
|
||||||
|
throw new CannotAcquireLockException("40001 still");
|
||||||
|
})).isInstanceOf(CannotAcquireLockException.class);
|
||||||
|
assertThat(calls.get()).isEqualTo(3);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void doesNotRetryOnOtherExceptions() {
|
||||||
|
AtomicInteger calls = new AtomicInteger();
|
||||||
|
assertThatThrownBy(() -> SerializableRetry.execute(() -> {
|
||||||
|
calls.incrementAndGet();
|
||||||
|
throw new IllegalStateException("nope");
|
||||||
|
})).isInstanceOf(IllegalStateException.class);
|
||||||
|
assertThat(calls.get()).isEqualTo(1);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user