Nail/IoT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P0-8: идемпотентность телеметрии (dedup по message_id)

Browse Source 
Envelope получил MessageID (UUID); таблица iot_telemetry — столбец message_id +
UNIQUE(device_id,message_id); батч-INSERT ON CONFLICT DO NOTHING. Дубли при
at-least-once повторе из SQS больше не вставляются. Пустой message_id → NULL
(без ложного дедупа).
This commit is contained in:
Repinoid
2026-10-07 07:26:50 +03:00
parent 3036572ccc
commit 38602b7560
5 changed files with 51 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HISTORY/2026-10-07-session-log.md
+13 -2
View File
@@ -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) — вне кода.
## Проверки
cmd/sqs-consumer/main.go
+1 -1
View File
@@ -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
}
internal/service/bridge/handler.go
+5
View File
@@ -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(),
})
}
}
internal/service/consumer/processor.go
+1 -1
View File
@@ -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
}
internal/storage/iotpg/iot_telemetry_store.go
+31 -14
View File
@@ -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