feat(idempotency): Idempotency-Key filter для exactly-once write API
OncePerRequestFilter перехватывает POST/PUT/PATCH с заголовком Idempotency-Key. Алгоритм: - sha-256(body) → request_hash - key найден с тем же hash → возврат cached response (status + body) - key с другим hash → 422 idempotency_conflict - key зарезервирован но без response → 409 idempotency_in_progress - новый key → reserve, запустить handler, кэшировать 2xx ответ Race-safety: reserve через PK saveAndFlush, DataIntegrityViolation поднимается до 409. retention 30 дней (cleanup job — будущая работа). Filter автоматически регистрируется через @Component и scanBasePackages в OrdinisApplication.
This commit is contained in:
Александр Зимин
+8
@@ -0,0 +1,8 @@
|
||||
package cloud.nstart.terravault.ordinis.restapi.idempotency;
|
||||
|
||||
/** HTTP 422 — клиент использует один и тот же Idempotency-Key для разных тел запроса. */
|
||||
public class IdempotencyConflictException extends RuntimeException {
|
||||
public IdempotencyConflictException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
+138
@@ -0,0 +1,138 @@
|
||||
package cloud.nstart.terravault.ordinis.restapi.idempotency;
|
||||
|
||||
import cloud.nstart.terravault.ordinis.domain.idempotency.IdempotencyKey;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import jakarta.servlet.FilterChain;
|
||||
import jakarta.servlet.ServletException;
|
||||
import jakarta.servlet.http.HttpServletRequest;
|
||||
import jakarta.servlet.http.HttpServletResponse;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.http.HttpHeaders;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.web.filter.OncePerRequestFilter;
|
||||
import org.springframework.web.util.ContentCachingRequestWrapper;
|
||||
import org.springframework.web.util.ContentCachingResponseWrapper;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
|
||||
/**
|
||||
* Servlet filter для exactly-once semantics на write API.
|
||||
*
|
||||
* <p>Срабатывает только на POST/PUT/PATCH (мутирующие методы) с заголовком
|
||||
* {@code Idempotency-Key}. На GET/HEAD/DELETE/OPTIONS — pass-through.
|
||||
*
|
||||
* <p>Логика:
|
||||
* <ol>
|
||||
* <li>Считает sha-256 хэш от request body.</li>
|
||||
* <li>Ищет ключ в БД. Если найден с тем же hash — возвращает cached
|
||||
* {@code response_status + response_body} без выполнения handler.</li>
|
||||
* <li>Если ключ есть с другим hash — 422 (конфликт тел запроса).</li>
|
||||
* <li>Если ключа нет — резервирует строку, передаёт управление вниз, после
|
||||
* успешного 2xx ответа сохраняет body в БД.</li>
|
||||
* </ol>
|
||||
*/
|
||||
@Component
|
||||
public class IdempotencyFilter extends OncePerRequestFilter {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(IdempotencyFilter.class);
|
||||
private static final String HEADER = "Idempotency-Key";
|
||||
private static final Set<String> MUTATING = Set.of("POST", "PUT", "PATCH");
|
||||
|
||||
private final IdempotencyService service;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public IdempotencyFilter(IdempotencyService service, ObjectMapper objectMapper) {
|
||||
this.service = service;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doFilterInternal(
|
||||
HttpServletRequest request, HttpServletResponse response, FilterChain chain)
|
||||
throws ServletException, IOException {
|
||||
|
||||
if (!MUTATING.contains(request.getMethod())) {
|
||||
chain.doFilter(request, response);
|
||||
return;
|
||||
}
|
||||
String key = request.getHeader(HEADER);
|
||||
if (key == null || key.isBlank()) {
|
||||
chain.doFilter(request, response);
|
||||
return;
|
||||
}
|
||||
if (key.length() > 128) {
|
||||
response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
|
||||
writeError(response, 400, "invalid_idempotency_key",
|
||||
"Idempotency-Key должен быть короче 128 символов");
|
||||
return;
|
||||
}
|
||||
|
||||
ContentCachingRequestWrapper wrappedReq = new ContentCachingRequestWrapper(request);
|
||||
// Need to read body to compute hash; do it after first content read. Spring caches via
|
||||
// ContentCachingRequestWrapper after handler reads, so we trigger the read manually.
|
||||
wrappedReq.getInputStream().readAllBytes(); // populate cache
|
||||
byte[] body = wrappedReq.getContentAsByteArray();
|
||||
String hash = service.hashRequest(body);
|
||||
|
||||
try {
|
||||
Optional<IdempotencyKey> cached = service.lookup(key, hash);
|
||||
if (cached.isPresent()) {
|
||||
IdempotencyKey ik = cached.get();
|
||||
if (ik.getResponseStatus() == null) {
|
||||
// зарезервирован, но обработка не завершилась — клиент должен retry
|
||||
response.setStatus(HttpServletResponse.SC_CONFLICT);
|
||||
writeError(response, 409, "idempotency_in_progress",
|
||||
"Запрос с Idempotency-Key '" + key + "' ещё обрабатывается");
|
||||
return;
|
||||
}
|
||||
log.debug("Idempotency hit для ключа '{}' → status={}", key, ik.getResponseStatus());
|
||||
response.setStatus(ik.getResponseStatus());
|
||||
response.setContentType(MediaType.APPLICATION_JSON_VALUE);
|
||||
if (ik.getResponseBody() != null) {
|
||||
byte[] bytes = objectMapper.writeValueAsBytes(ik.getResponseBody());
|
||||
response.setContentLength(bytes.length);
|
||||
response.getOutputStream().write(bytes);
|
||||
}
|
||||
return;
|
||||
}
|
||||
service.reserve(key, hash);
|
||||
} catch (IdempotencyConflictException e) {
|
||||
response.setStatus(422);
|
||||
writeError(response, 422, "idempotency_conflict", e.getMessage());
|
||||
return;
|
||||
} catch (IdempotencyInProgressException e) {
|
||||
response.setStatus(HttpServletResponse.SC_CONFLICT);
|
||||
writeError(response, 409, "idempotency_in_progress", e.getMessage());
|
||||
return;
|
||||
}
|
||||
|
||||
ContentCachingResponseWrapper wrappedResp = new ContentCachingResponseWrapper(response);
|
||||
chain.doFilter(wrappedReq, wrappedResp);
|
||||
|
||||
int status = wrappedResp.getStatus();
|
||||
byte[] respBody = wrappedResp.getContentAsByteArray();
|
||||
// Кэшируем только успешные ответы; ошибки могут быть транзиентными — клиент retry'ит.
|
||||
if (status >= 200 && status < 300) {
|
||||
try {
|
||||
service.saveResponse(key, status, respBody);
|
||||
} catch (Exception e) {
|
||||
log.warn("Не удалось сохранить idempotent response для ключа '{}': {}", key, e.getMessage());
|
||||
}
|
||||
}
|
||||
wrappedResp.copyBodyToResponse();
|
||||
}
|
||||
|
||||
private void writeError(HttpServletResponse response, int status, String code, String message)
|
||||
throws IOException {
|
||||
response.setContentType(MediaType.APPLICATION_JSON_VALUE);
|
||||
response.setHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
|
||||
String json = "{\"status\":" + status
|
||||
+ ",\"code\":\"" + code + "\""
|
||||
+ ",\"message\":\"" + message.replace("\"", "\\\"") + "\"}";
|
||||
response.getOutputStream().write(json.getBytes());
|
||||
}
|
||||
}
|
||||
+8
@@ -0,0 +1,8 @@
|
||||
package cloud.nstart.terravault.ordinis.restapi.idempotency;
|
||||
|
||||
/** HTTP 409 — параллельный запрос с тем же ключом ещё обрабатывается. */
|
||||
public class IdempotencyInProgressException extends RuntimeException {
|
||||
public IdempotencyInProgressException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
+117
@@ -0,0 +1,117 @@
|
||||
package cloud.nstart.terravault.ordinis.restapi.idempotency;
|
||||
|
||||
import cloud.nstart.terravault.ordinis.domain.idempotency.IdempotencyKey;
|
||||
import cloud.nstart.terravault.ordinis.domain.idempotency.IdempotencyKeyRepository;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.dao.DataIntegrityViolationException;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.time.OffsetDateTime;
|
||||
import java.util.HexFormat;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* Хранение и lookup для Idempotency-Key.
|
||||
*
|
||||
* <p>Контракт:
|
||||
* <ul>
|
||||
* <li>{@link #lookup(String, String)} — возвращает существующий ключ если он уже
|
||||
* видел тот же request hash. Если hash не совпадает — кидает
|
||||
* {@link IdempotencyConflictException} (HTTP 422 у клиента).</li>
|
||||
* <li>{@link #reserve(String, String)} — пытается атомарно зарегистрировать
|
||||
* новый ключ. Бросает {@link IdempotencyConflictException} если ключ уже
|
||||
* занят чужим hash (race-safe через unique PK constraint).</li>
|
||||
* <li>{@link #saveResponse(String, int, JsonNode)} — после успешной обработки
|
||||
* сохраняет response для повтора.</li>
|
||||
* </ul>
|
||||
*/
|
||||
@Service
|
||||
public class IdempotencyService {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(IdempotencyService.class);
|
||||
|
||||
/** Хранилище: 30 дней (соответствует комментарию в migration 0006). */
|
||||
private static final long RETENTION_DAYS = 30;
|
||||
|
||||
private final IdempotencyKeyRepository repository;
|
||||
private final ObjectMapper objectMapper;
|
||||
|
||||
public IdempotencyService(IdempotencyKeyRepository repository, ObjectMapper objectMapper) {
|
||||
this.repository = repository;
|
||||
this.objectMapper = objectMapper;
|
||||
}
|
||||
|
||||
/** Sha-256 hex от тела запроса; идентичен только для байт-в-байт идентичных тел. */
|
||||
public String hashRequest(byte[] body) {
|
||||
try {
|
||||
MessageDigest md = MessageDigest.getInstance("SHA-256");
|
||||
return HexFormat.of().formatHex(md.digest(body == null ? new byte[0] : body));
|
||||
} catch (NoSuchAlgorithmException e) {
|
||||
throw new IllegalStateException("SHA-256 not available", e);
|
||||
}
|
||||
}
|
||||
|
||||
@Transactional(readOnly = true)
|
||||
public Optional<IdempotencyKey> lookup(String key, String requestHash) {
|
||||
Optional<IdempotencyKey> existing = repository.findById(key);
|
||||
if (existing.isEmpty()) {
|
||||
return Optional.empty();
|
||||
}
|
||||
IdempotencyKey ik = existing.get();
|
||||
if (!ik.getRequestHash().equals(requestHash)) {
|
||||
throw new IdempotencyConflictException(
|
||||
"Idempotency-Key '" + key + "' уже использовался с другим телом запроса");
|
||||
}
|
||||
return existing;
|
||||
}
|
||||
|
||||
/**
|
||||
* Резервирует ключ перед обработкой. Если запись уже создана — поднимает 422
|
||||
* (клиент видит, что мы только что начали обрабатывать тот же ключ; обычно это
|
||||
* следствие гонки double-submit, и второй запрос должен retry'ить).
|
||||
*/
|
||||
@Transactional
|
||||
public void reserve(String key, String requestHash) {
|
||||
OffsetDateTime expires = OffsetDateTime.now().plusDays(RETENTION_DAYS);
|
||||
IdempotencyKey reserved = new IdempotencyKey(key, requestHash, expires);
|
||||
try {
|
||||
repository.saveAndFlush(reserved);
|
||||
} catch (DataIntegrityViolationException e) {
|
||||
// Кто-то параллельный успел зарезервировать тот же ключ.
|
||||
Optional<IdempotencyKey> existing = repository.findById(key);
|
||||
if (existing.isPresent() && existing.get().getRequestHash().equals(requestHash)) {
|
||||
// То же тело — клиент может retry'ить, мы вернём 409 чтобы он понял что обработка ещё идёт.
|
||||
throw new IdempotencyInProgressException(
|
||||
"Запрос с Idempotency-Key '" + key + "' ещё обрабатывается");
|
||||
}
|
||||
throw new IdempotencyConflictException(
|
||||
"Idempotency-Key '" + key + "' уже использовался с другим телом запроса");
|
||||
}
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public void saveResponse(String key, int status, byte[] body) {
|
||||
IdempotencyKey ik = repository.findById(key).orElse(null);
|
||||
if (ik == null) {
|
||||
log.warn("saveResponse для отсутствующего ключа '{}', пропускаем", key);
|
||||
return;
|
||||
}
|
||||
JsonNode parsed = null;
|
||||
if (body != null && body.length > 0) {
|
||||
try {
|
||||
parsed = objectMapper.readTree(body);
|
||||
} catch (Exception e) {
|
||||
log.debug("Тело ответа для ключа '{}' не JSON, не кэшируем", key);
|
||||
}
|
||||
}
|
||||
ik.setResponse(status, parsed);
|
||||
repository.save(ik);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user