diff --git a/HISTORY/2026-10-07-session-log.md b/HISTORY/2026-10-07-session-log.md index f57eea5..8ce3340 100644 --- a/HISTORY/2026-10-07-session-log.md +++ b/HISTORY/2026-10-07-session-log.md @@ -50,9 +50,20 @@ - `cmd/iot-service/main.go` — bridge/consumer поднимают/опускают свои флаги; если горутина завершилась (ошибка/падение), `/ready` вернёт 503. +### P0-8 — идемпотентность телеметрии (dedup по message_id) +- `internal/service/bridge/handler.go`: в `TelemetryEnvelope` добавлен `MessageID` + (UUID на каждое MQTT-сообщение). +- `internal/service/consumer/processor.go`: `MessageID` передаётся в `InsertTelemetry`. +- `internal/storage/iotpg/iot_telemetry_store.go`: в таблицу добавлен столбец + `message_id` + UNIQUE INDEX `(device_id, message_id)`; батч-INSERT теперь + `ON CONFLICT DO NOTHING`. Пустой message_id → NULL (NULL в unique-индексе не + конфликтует → без ложного дедупа). При at-least-once повторе из SQS дубликат + не вставляется. +- Легаси `cmd/sqs-consumer`: вызов обновлён (`""`), файл помечен к удалению (P2-14). + ## Осталось (не сделано) -- P0-3 часть 1 (хэш паролей + EMQX), P0-8 (идемпотентность), P1-11/P1-12 - (тесты/CI), P2-14 (вычистить легаси `internal/api/`). +- P0-3 часть 1 (хэш паролей + EMQX), P1-11/P1-12 (тесты/CI), P2-14 + (вычистить легаси `internal/api/`, `cmd/sqs-consumer`). - Платформенные проблемы (wss ~150с, MSS/MTU) — вне кода. ## Проверки diff --git a/cmd/sqs-consumer/main.go b/cmd/sqs-consumer/main.go index f9a0f7d..04b917d 100644 --- a/cmd/sqs-consumer/main.go +++ b/cmd/sqs-consumer/main.go @@ -160,7 +160,7 @@ func processSQSTelemetry(ctx context.Context, body string, store *iotpg.IoTPostg return err } - if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.Payload); err != nil { + if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, "", envelope.Payload); err != nil { return err } diff --git a/internal/service/bridge/handler.go b/internal/service/bridge/handler.go index fa8d164..a98dd05 100644 --- a/internal/service/bridge/handler.go +++ b/internal/service/bridge/handler.go @@ -11,6 +11,7 @@ import ( "time" mqtt "github.com/eclipse/paho.mqtt.golang" + "github.com/google/uuid" ) // TelemetryEnvelope — сообщение, перекладываемое в SQS (разбирается consumer). @@ -20,6 +21,9 @@ type TelemetryEnvelope struct { Topic string `json:"topic"` Payload json.RawMessage `json:"payload"` ReceivedAt string `json:"received_at"` + // MessageID — уникальный id сообщения. Нужен для идемпотентности: SQS + // at-least-once может доставить дубликат — consumer дедуплицирует по нему. + MessageID string `json:"message_id"` } // newMessageHandler возвращает обработчик MQTT-сообщений: @@ -51,6 +55,7 @@ func newMessageHandler(dispatcher *sqsDispatcher, log *slog.Logger) mqtt.Message Topic: topic, Payload: rawPayload, ReceivedAt: time.Now().UTC().Format(time.RFC3339), + MessageID: uuid.NewString(), }) } } diff --git a/internal/service/consumer/processor.go b/internal/service/consumer/processor.go index f1ca974..3e6f6af 100644 --- a/internal/service/consumer/processor.go +++ b/internal/service/consumer/processor.go @@ -23,7 +23,7 @@ func processTelemetry(ctx context.Context, body string, store *iotpg.IoTPostgres if err := store.EnsureTenantDB(ctx, envelope.Namespace); err != nil { return err } - if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.Payload); err != nil { + if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.MessageID, envelope.Payload); err != nil { return err } diff --git a/internal/storage/iotpg/iot_telemetry_store.go b/internal/storage/iotpg/iot_telemetry_store.go index 3ecf29d..3a58c49 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -50,8 +50,9 @@ type IoTPostgresStore struct { // telemetryInsert — строка для батч-вставки. type telemetryInsert struct { - deviceID string - payload []byte + deviceID string + messageID string + payload []byte } // telemetryBatchSize — размер батча перед синхронным flush. @@ -157,17 +158,25 @@ func (s *IoTPostgresStore) flushTenant(ns string, rows []telemetryInsert) error if err != nil { return err } - // INSERT INTO iot_telemetry (device_id, payload) VALUES ($1,$2),($3,$4),... + // INSERT INTO iot_telemetry (device_id, payload, message_id) VALUES ($1,$2,$3),... + // ON CONFLICT DO NOTHING — дедуп по (device_id, message_id) при at-least-once + // повторе из SQS (P0-8). Пустой message_id вставляется как NULL (NULL в + // уникальном индексе не конфликтует → без ложного дедупа). var sb strings.Builder - sb.WriteString("INSERT INTO iot_telemetry (device_id, payload) VALUES ") - args := make([]any, 0, len(rows)*2) + sb.WriteString("INSERT INTO iot_telemetry (device_id, payload, message_id) VALUES ") + args := make([]any, 0, len(rows)*3) for i, r := range rows { if i > 0 { sb.WriteString(",") } - sb.WriteString(fmt.Sprintf("($%d,$%d)", i*2+1, i*2+2)) - args = append(args, r.deviceID, r.payload) + sb.WriteString(fmt.Sprintf("($%d,$%d,$%d)", i*3+1, i*3+2, i*3+3)) + var mid any + if r.messageID != "" { + mid = r.messageID + } + args = append(args, r.deviceID, r.payload, mid) } + sb.WriteString(" ON CONFLICT DO NOTHING") _, err = tenantDB.ExecContext(ctx, sb.String(), args...) return err } @@ -270,13 +279,18 @@ func (s *IoTPostgresStore) EnsureTenantDB(ctx context.Context, namespace string) } _, err = tenantDB.ExecContext(ctx, ` CREATE TABLE IF NOT EXISTS iot_telemetry ( -id BIGSERIAL PRIMARY KEY, -device_id TEXT NOT NULL, -ts TIMESTAMPTZ NOT NULL DEFAULT now(), -payload JSONB NOT NULL +id BIGSERIAL PRIMARY KEY, +device_id TEXT NOT NULL, +message_id TEXT, +ts TIMESTAMPTZ NOT NULL DEFAULT now(), +payload JSONB NOT NULL ); CREATE INDEX IF NOT EXISTS idx_iot_telemetry_device_ts ON iot_telemetry (device_id, ts DESC); +-- Миграция существующих таблиц + уникальный ключ дедупликации (P0-8). +ALTER TABLE iot_telemetry ADD COLUMN IF NOT EXISTS message_id TEXT; +CREATE UNIQUE INDEX IF NOT EXISTS uniq_iot_telemetry_device_msg +ON iot_telemetry (device_id, message_id); `) if err != nil { return err @@ -288,11 +302,14 @@ ON iot_telemetry (device_id, ts DESC); // InsertTelemetry ставит строку в батч-буфер тенанта. // При накоплении telemetryBatchSize строк батч пишется синхронно (ошибка // возвращается вызывающему); неполные батчи дописывает фоновый flusher. -func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID string, payload json.RawMessage) error { +// messageID — уникальный id сообщения для дедупликации (может быть пустым → +// тогда строка не участвует в дедупе). +func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID, messageID string, payload json.RawMessage) error { s.mu.Lock() s.batches[namespace] = append(s.batches[namespace], telemetryInsert{ - deviceID: deviceID, - payload: append([]byte(nil), payload...), + deviceID: deviceID, + messageID: messageID, + payload: append([]byte(nil), payload...), }) full := len(s.batches[namespace]) >= telemetryBatchSize var rows []telemetryInsert