P0-7: подтверждение из SQS только после успешного flush телеметрии в БД
InsertTelemetry лишь буферизует и возвращает nil до фактической записи. Consumer удалял сообщение сразу после processTelemetry == nil — краш/flush-ошибка в этом окне = безвозвратная потеря. Теперь: копим receipt handles, вызываем store.Flush, и только при успехе DeleteMessage; при ошибке flush — не удаляем (сообщения вернутся после visibility timeout). Добавлен iotpg.Flush().
This commit is contained in:
@@ -141,6 +141,14 @@ func (s *IoTPostgresStore) flushAll(reportErr bool) error {
|
||||
return firstErr
|
||||
}
|
||||
|
||||
// Flush — принудительно синхронно пишет все накопленные батчи телеметрии в БД и
|
||||
// возвращает первую ошибку. Нужен consumer'у, чтобы подтверждать (удалять из SQS)
|
||||
// ТОЛЬКО фактически записанные данные: без этого краш/flush-ошибка после
|
||||
// InsertTelemetry (которая лишь буферизует и возвращает nil) = потеря сообщений.
|
||||
func (s *IoTPostgresStore) Flush(ctx context.Context) error {
|
||||
return s.flushAll(true)
|
||||
}
|
||||
|
||||
// flushTenant — INSERT батча строк одного тенанта (multi-VALUES).
|
||||
func (s *IoTPostgresStore) flushTenant(ns string, rows []telemetryInsert) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
|
||||
Reference in New Issue
Block a user