fix: review findings (async SQS dispatcher, backoff, batch insert, shutdown order, pagination, HMAC) + loadtest fixes; v0.1.6

2026-08-16 18:24:09 +04:00
parent 090cedca67
commit 3fedf9317c
27 changed files with 581 additions and 122 deletions
@@ -37,9 +37,29 @@ type IoTPostgresStore struct {
 	adminDB  *sql.DB
 	adminDSN string
 	tenants  sync.Map
+	tenantMu sync.Map // namespace → *sync.Mutex (дедупликация открытия)
+	ensured  sync.Map // namespace → struct{} (EnsureTenantDB уже выполнен)
 	log      *slog.Logger
+
+	// батчинг вставок (фикс MEDIUM ревью 2026-08-16: 1000 msg/s = 1000 INSERT)
+	mu        sync.Mutex
+	batches   map[string][]telemetryInsert
+	stopFlush chan struct{}
+	flushWG   sync.WaitGroup
 }

+// telemetryInsert — строка для батч-вставки.
+type telemetryInsert struct {
+	deviceID string
+	payload  []byte
+}
+
+// telemetryBatchSize — размер батча перед синхронным flush.
+const telemetryBatchSize = 100
+
+// telemetryFlushInterval — период фонового flush неполных батчей.
+const telemetryFlushInterval = 200 * time.Millisecond
+
 // TelemetryRow — одна запись телеметрии из таблицы iot_telemetry.
 type TelemetryRow struct {
 	ID       int64           `json:"id"`
@@ -60,19 +80,90 @@ func New(adminDSN string, log *slog.Logger) (*IoTPostgresStore, error) {
 		db.Close()
 		return nil, fmt.Errorf("iotpg: ping admin DB: %w", err)
 	}
-	db.SetMaxOpenConns(5)
+	db.SetMaxOpenConns(10)
 	db.SetMaxIdleConns(2)
 	db.SetConnMaxLifetime(5 * time.Minute)

-	store := &IoTPostgresStore{adminDB: db, adminDSN: adminDSN, log: log}
+	store := &IoTPostgresStore{
+		adminDB:   db,
+		adminDSN:  adminDSN,
+		batches:   make(map[string][]telemetryInsert),
+		stopFlush: make(chan struct{}),
+		log:       log,
+	}
 	if err := store.initManagementSchema(ctx); err != nil {
 		db.Close()
 		return nil, fmt.Errorf("iotpg: init management schema: %w", err)
 	}
+	store.startFlusher()
 	log.Info("iotpg: connected to IoT Postgres management DB")
 	return store, nil
 }

+// startFlusher — фоновая горутина периодического flush неполных батчей.
+func (s *IoTPostgresStore) startFlusher() {
+	s.flushWG.Add(1)
+	go func() {
+		defer s.flushWG.Done()
+		t := time.NewTicker(telemetryFlushInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-s.stopFlush:
+				return
+			case <-t.C:
+				s.flushAll(false)
+			}
+		}
+	}()
+}
+
+// flushAll — flush всех накопленных батчей. reportErr=false: только лог.
+func (s *IoTPostgresStore) flushAll(reportErr bool) error {
+	s.mu.Lock()
+	if len(s.batches) == 0 {
+		s.mu.Unlock()
+		return nil
+	}
+	batches := s.batches
+	s.batches = make(map[string][]telemetryInsert)
+	s.mu.Unlock()
+
+	var firstErr error
+	for ns, rows := range batches {
+		if err := s.flushTenant(ns, rows); err != nil {
+			s.log.Error("iotpg: flush batch", "namespace", ns, "rows", len(rows), "err", err)
+			if firstErr == nil {
+				firstErr = err
+			}
+		}
+	}
+	return firstErr
+}
+
+// flushTenant — INSERT батча строк одного тенанта (multi-VALUES).
+func (s *IoTPostgresStore) flushTenant(ns string, rows []telemetryInsert) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	tenantDB, err := s.getTenantDB(ctx, ns)
+	if err != nil {
+		return err
+	}
+	// INSERT INTO iot_telemetry (device_id, payload) VALUES ($1,$2),($3,$4),...
+	var sb strings.Builder
+	sb.WriteString("INSERT INTO iot_telemetry (device_id, payload) VALUES ")
+	args := make([]any, 0, len(rows)*2)
+	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)
+	}
+	_, err = tenantDB.ExecContext(ctx, sb.String(), args...)
+	return err
+}
+
 // NewFromEnv создаёт store из env var IOT_PG_DSN.
 // Возвращает (nil, nil) если переменная не задана — IoT Postgres опционален.
 func NewFromEnv(log *slog.Logger) (*IoTPostgresStore, error) {
@@ -97,9 +188,12 @@ created_at  TIMESTAMPTZ DEFAULT now()
 }

 // EnsureTenantDB создаёт DATABASE, USER и таблицу iot_telemetry для namespace.
-// Идемпотентен — повторный вызов безопасен.
-// Вызывается mqtt-bridge при первом сообщении от нового tenant.
+// Идемпотентен — повторный вызов безопасен (кэш ensured пропускает проверки).
 func (s *IoTPostgresStore) EnsureTenantDB(ctx context.Context, namespace string) error {
+	if _, ok := s.ensured.Load(namespace); ok {
+		return nil
+	}
+
 	dbName := tenantDBName(namespace)
 	userName := dbName

@@ -114,28 +208,39 @@ func (s *IoTPostgresStore) EnsureTenantDB(ctx context.Context, namespace string)
 	if !exists {
 		password := uuid.New().String()

-		// CREATE USER через DO block — pg не поддерживает CREATE USER IF NOT EXISTS
-		_, err = s.adminDB.ExecContext(ctx, fmt.Sprintf(
-			`DO $$ BEGIN
-                IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = '%s') THEN
-                    CREATE USER %s WITH PASSWORD '%s';
-                END IF;
-            END $$`, userName, userName, password,
-		))
+		// Роль создаём только если её нет (DO-блоки НЕ принимают параметры —
+		// прецедент 2026-08-16: "got 2 parameters but the statement requires 0").
+		var roleExists bool
+		err = s.adminDB.QueryRowContext(ctx,
+			`SELECT EXISTS(SELECT 1 FROM pg_roles WHERE rolname = $1)`, userName,
+		).Scan(&roleExists)
 		if err != nil {
-			return fmt.Errorf("iotpg: create user %s: %w", userName, err)
+			return fmt.Errorf("iotpg: check role %s: %w", userName, err)
+		}
+		if !roleExists {
+			_, err = s.adminDB.ExecContext(ctx,
+				`CREATE USER `+pq.QuoteIdentifier(userName)+` WITH PASSWORD `+pq.QuoteLiteral(password),
+			)
+			if err != nil {
+				var pqErr *pq.Error
+				if errors.As(err, &pqErr) && pqErr.Code == "42710" { // duplicate_object
+					// гонка: роль создал параллельный вызов — ок
+				} else {
+					return fmt.Errorf("iotpg: create user %s: %w", userName, err)
+				}
+			}
 		}

 		// PG15+: GRANT role TO current_user перед CREATE DATABASE ... OWNER
 		if _, err = s.adminDB.ExecContext(ctx,
-			fmt.Sprintf(`GRANT %s TO CURRENT_USER`, userName),
+			`GRANT `+pq.QuoteIdentifier(userName)+` TO CURRENT_USER`,
 		); err != nil {
 			return fmt.Errorf("iotpg: grant role %s: %w", userName, err)
 		}

 		// CREATE DATABASE нельзя в транзакции
 		if _, err = s.adminDB.ExecContext(ctx,
-			fmt.Sprintf(`CREATE DATABASE %s OWNER %s`, dbName, userName),
+			`CREATE DATABASE `+pq.QuoteIdentifier(dbName)+` OWNER `+pq.QuoteIdentifier(userName),
 		); err != nil {
 			return fmt.Errorf("iotpg: create database %s: %w", dbName, err)
 		}
@@ -165,32 +270,50 @@ payload   JSONB       NOT NULL
 CREATE INDEX IF NOT EXISTS idx_iot_telemetry_device_ts
 ON iot_telemetry (device_id, ts DESC);
 `)
-	return err
+	if err != nil {
+		return err
+	}
+	s.ensured.Store(namespace, struct{}{})
+	return nil
 }

-// InsertTelemetry записывает строку телеметрии в tenant DB.
+// InsertTelemetry ставит строку в батч-буфер тенанта.
+// При накоплении telemetryBatchSize строк батч пишется синхронно (ошибка
+// возвращается вызывающему); неполные батчи дописывает фоновый flusher.
 func (s *IoTPostgresStore) InsertTelemetry(ctx context.Context, namespace, deviceID string, payload json.RawMessage) error {
-	tenantDB, err := s.getTenantDB(ctx, namespace)
-	if err != nil {
-		return fmt.Errorf("iotpg: get tenant DB for insert: %w", err)
+	s.mu.Lock()
+	s.batches[namespace] = append(s.batches[namespace], telemetryInsert{
+		deviceID: deviceID,
+		payload:  append([]byte(nil), payload...),
+	})
+	full := len(s.batches[namespace]) >= telemetryBatchSize
+	var rows []telemetryInsert
+	if full {
+		rows = s.batches[namespace]
+		delete(s.batches, namespace)
 	}
-	_, err = tenantDB.ExecContext(ctx,
-		`INSERT INTO iot_telemetry (device_id, payload) VALUES ($1, $2)`,
-		deviceID, []byte(payload),
-	)
-	return err
+	s.mu.Unlock()
+
+	if full {
+		return s.flushTenant(namespace, rows)
+	}
+	return nil
 }

 // QueryTelemetry читает телеметрию из tenant DB (ts DESC).
-// deviceID — фильтр (пустая строка = все устройства). limit — max записей (50..1000).
-// Если tenant DB не существует (данных ещё нет) — возвращает пустой срез без ошибки.
-func (s *IoTPostgresStore) QueryTelemetry(ctx context.Context, namespace, deviceID string, limit int) ([]TelemetryRow, error) {
+// deviceID — фильтр (пустая строка = все устройства). limit — max записей
+// (50..1000), offset — смещение для пагинации.
+// Если tenant DB не существует (данных ещё нет) — пустой срез без ошибки.
+func (s *IoTPostgresStore) QueryTelemetry(ctx context.Context, namespace, deviceID string, limit, offset int) ([]TelemetryRow, error) {
 	if limit <= 0 {
 		limit = 50
 	}
 	if limit > 1000 {
 		limit = 1000
 	}
+	if offset < 0 {
+		offset = 0
+	}
 	tenantDB, err := s.getTenantDB(ctx, namespace)
 	if err != nil {
 		// Если DB не существует — тенант ещё не отправлял данные, это нормально
@@ -204,14 +327,14 @@ func (s *IoTPostgresStore) QueryTelemetry(ctx context.Context, namespace, device
 	if deviceID != "" {
 		rows, err = tenantDB.QueryContext(ctx,
 			`SELECT id, device_id, ts, payload FROM iot_telemetry
- WHERE device_id = $1 ORDER BY ts DESC LIMIT $2`,
-			deviceID, limit,
+ WHERE device_id = $1 ORDER BY ts DESC LIMIT $2 OFFSET $3`,
+			deviceID, limit, offset,
 		)
 	} else {
 		rows, err = tenantDB.QueryContext(ctx,
 			`SELECT id, device_id, ts, payload FROM iot_telemetry
- ORDER BY ts DESC LIMIT $1`,
-			limit,
+ ORDER BY ts DESC LIMIT $1 OFFSET $2`,
+			limit, offset,
 		)
 	}
 	if err != nil {
@@ -308,7 +431,6 @@ FROM iot_telemetry`).Scan(&stats.Total, &stats.Last1h, &stats.Last24h)
 		latestRows, err := tenantDB.QueryContext(ctx,
 			`SELECT id, device_id, ts, payload FROM iot_telemetry ORDER BY ts DESC LIMIT 5`)
 		if err == nil {
-			defer latestRows.Close()
 			for latestRows.Next() {
 				var r TelemetryRow
 				var rawPayload []byte
@@ -317,6 +439,7 @@ FROM iot_telemetry`).Scan(&stats.Total, &stats.Last1h, &stats.Last24h)
 					stats.Latest = append(stats.Latest, r)
 				}
 			}
+			latestRows.Close() // закрываем сразу, не defer в цикле
 		}

 		result.Tenants = append(result.Tenants, stats)
@@ -336,8 +459,11 @@ func isDBNotExistErr(err error) bool {
 	return false
 }

-// Close закрывает все подключения (admin + tenant кэш).
+// Close останавливает flusher, дописывает остаток и закрывает все подключения.
 func (s *IoTPostgresStore) Close() error {
+	close(s.stopFlush)
+	s.flushWG.Wait()
+	_ = s.flushAll(true)
 	s.tenants.Range(func(_, value any) bool {
 		if db, ok := value.(*sql.DB); ok {
 			db.Close()
@@ -348,7 +474,16 @@ func (s *IoTPostgresStore) Close() error {
 }

 // getTenantDB возвращает *sql.DB для tenant DB из кэша или открывает новый.
+// Открытие дедуплицируется per-namespace мьютексом (гонка Load→LoadOrStore
+// из ревью 2026-08-16).
 func (s *IoTPostgresStore) getTenantDB(ctx context.Context, namespace string) (*sql.DB, error) {
+	if cached, ok := s.tenants.Load(namespace); ok {
+		return cached.(*sql.DB), nil
+	}
+	m, _ := s.tenantMu.LoadOrStore(namespace, &sync.Mutex{})
+	mu := m.(*sync.Mutex)
+	mu.Lock()
+	defer mu.Unlock()
 	if cached, ok := s.tenants.Load(namespace); ok {
 		return cached.(*sql.DB), nil
 	}