From e8873fa24d5109ab81966855ea075538e85063f5 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Wed, 7 Oct 2026 07:47:10 +0300 Subject: [PATCH] =?UTF-8?q?fix(telemetry):=20ts=20=3D=20=D0=B2=D1=80=D0=B5?= =?UTF-8?q?=D0=BC=D1=8F=20=D0=BF=D1=80=D0=B8=D1=91=D0=BC=D0=B0=20(Received?= =?UTF-8?q?At),=20=D0=B0=20=D0=BD=D0=B5=20=D0=BC=D0=BE=D0=BC=D0=B5=D0=BD?= =?UTF-8?q?=D1=82=20INSERT?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Раньше ts проставлялся now() в БД при INSERT → при батчинге и параллельных записях порядок телеметрии (ORDER BY ts DESC) ломался. Теперь в ts идёт ReceivedAt из envelope (серверное время приёма, ставит bridge); битый/пустой received_at → zero → стор подставляет now(). Схема не меняется (используется существующая колонка ts). Обновлены callers: consumer/processor.go и legacy cmd/sqs-consumer. --- cmd/sqs-consumer/main.go | 6 ++++- internal/service/consumer/processor.go | 9 ++++++- internal/storage/iotpg/iot_telemetry_store.go | 26 ++++++++++++------- 3 files changed, 30 insertions(+), 11 deletions(-) diff --git a/cmd/sqs-consumer/main.go b/cmd/sqs-consumer/main.go index 04b917d..ad41f41 100644 --- a/cmd/sqs-consumer/main.go +++ b/cmd/sqs-consumer/main.go @@ -160,7 +160,11 @@ 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 { + receivedAt, perr := time.Parse(time.RFC3339, envelope.ReceivedAt) + if perr != nil { + receivedAt = time.Time{} + } + if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, "", receivedAt, envelope.Payload); err != nil { return err } diff --git a/internal/service/consumer/processor.go b/internal/service/consumer/processor.go index 32a0f39..254947b 100644 --- a/internal/service/consumer/processor.go +++ b/internal/service/consumer/processor.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "log/slog" + "time" "gitea.services.ngcloud.ru/Nail/IoT/internal/service/bridge" "gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg" @@ -29,7 +30,13 @@ 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.MessageID, envelope.Payload); err != nil { + // Время приёма сохраняем в ts (порядок телеметрии). Битый/пустой received_at + // → zero, стор подставит now(). + receivedAt, perr := time.Parse(time.RFC3339, envelope.ReceivedAt) + if perr != nil { + receivedAt = time.Time{} + } + if err := store.InsertTelemetry(ctx, envelope.Namespace, envelope.DeviceID, envelope.MessageID, receivedAt, 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 2d0dcbb..43e7a9d 100644 --- a/internal/storage/iotpg/iot_telemetry_store.go +++ b/internal/storage/iotpg/iot_telemetry_store.go @@ -52,6 +52,7 @@ type IoTPostgresStore struct { type telemetryInsert struct { deviceID string messageID string + ts time.Time payload []byte } @@ -172,23 +173,28 @@ func (s *IoTPostgresStore) flushTenant(ctx context.Context, ns string, rows []te if err != nil { return err } - // 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 в - // уникальном индексе не конфликтует → без ложного дедупа). + // INSERT INTO iot_telemetry (device_id, payload, message_id, ts) VALUES ($1,$2,$3,$4),... + // ts — ВРЕМЯ ПРИЁМА (bridge ReceivedAt), а не момент INSERT: иначе батчинг и + // параллельные записи ломают порядок телеметрии (ORDER BY ts DESC). + // Zero → now() (fallback для legacy-сообщений без received_at). + // ON CONFLICT DO NOTHING — дедуп по (device_id, message_id) (P0-8). var sb strings.Builder - sb.WriteString("INSERT INTO iot_telemetry (device_id, payload, message_id) VALUES ") - args := make([]any, 0, len(rows)*3) + sb.WriteString("INSERT INTO iot_telemetry (device_id, payload, message_id, ts) VALUES ") + args := make([]any, 0, len(rows)*4) for i, r := range rows { if i > 0 { sb.WriteString(",") } - sb.WriteString(fmt.Sprintf("($%d,$%d,$%d)", i*3+1, i*3+2, i*3+3)) + sb.WriteString(fmt.Sprintf("($%d,$%d,$%d,$%d)", i*4+1, i*4+2, i*4+3, i*4+4)) var mid any if r.messageID != "" { mid = r.messageID } - args = append(args, r.deviceID, r.payload, mid) + ts := r.ts + if ts.IsZero() { + ts = time.Now().UTC() + } + args = append(args, r.deviceID, r.payload, mid, ts) } sb.WriteString(" ON CONFLICT DO NOTHING") _, err = tenantDB.ExecContext(ctx, sb.String(), args...) @@ -318,11 +324,13 @@ ON iot_telemetry (device_id, message_id); // возвращается вызывающему); неполные батчи дописывает фоновый flusher. // messageID — уникальный id сообщения для дедупликации (может быть пустым → // тогда строка не участвует в дедупе). -func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID, messageID string, payload json.RawMessage) error { +// receivedAt — время приёма сообщения (для колонки ts); zero → стор подставит now(). +func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID, messageID string, receivedAt time.Time, payload json.RawMessage) error { s.mu.Lock() s.batches[namespace] = append(s.batches[namespace], telemetryInsert{ deviceID: deviceID, messageID: messageID, + ts: receivedAt, payload: append([]byte(nil), payload...), }) full := len(s.batches[namespace]) >= telemetryBatchSize