отработаны находки ревью кода
Триаж свёл 62 сырые находки девяти проходов к 33 причинам: 3 блокера, 4 «сейчас», 2 развилки. Все закрыты регрессионными тестами. - схема точки сна определяется по самой точке, а не по индексу в исходном массиве: одна пропущенная точка меняла эпизод и сводку местами - доставка сворачивается одной транзакцией: частичное состояние было недетерминированным (восемь прогонов — семь состояний) - граница размера на распакованном теле: 400 КиБ gzip разворачивались в 400 МиБ мимо max_body_mb - столкновение — расхождение канонических форм, а не байтов; WARN с координатами объекта; payload без HTML-экранирования - выравнивание по местной метке: получасовые зоны уводили часовую выгрузку в minute - доставка из одних суточных сводок больше не отвергается целиком - единицы не переписываются молча; счётчик считает сохранённые точки - все выходы Fold логируются, исход пишется на переживающем отмену контексте Четыре развилки вынесены блокерами в беклог.
This commit is contained in:
+162
-73
@@ -46,39 +46,111 @@ type Bucket struct {
|
||||
// это единственное наблюдение, по которому можно судить, работает ли правило
|
||||
// разрешения столкновений.
|
||||
type MergeStats struct {
|
||||
Buckets int
|
||||
Points int
|
||||
Buckets int
|
||||
// Stored — сколько точек ДОБАВИЛОСЬ в объекты. Не то же, что число точек в
|
||||
// доставке: точные повторы схлопываются, и счётчик присланных завышал бы
|
||||
// содержимое витрины ровно тогда, когда точки начнут теряться.
|
||||
Stored int
|
||||
Unchanged int
|
||||
Overwrites int
|
||||
SealedHits int
|
||||
// UnitsConflicts — объекты, куда точки приехали в единицах, отличных от
|
||||
// сохранённых. Молчаливая замена недопустима: внутри точки единиц нет, и
|
||||
// у ранних точек не остаётся ничего, по чему их единицы восстановимы.
|
||||
UnitsConflicts int
|
||||
// Collisions — координаты первых столкновений, для записи в лог. Без них
|
||||
// счётчик перезаписей не говорит, какая метрика и какой час пострадали.
|
||||
Collisions []Collision
|
||||
}
|
||||
|
||||
// Collision — координаты объекта, где содержимое точки было перезаписано.
|
||||
// Значений точек не несёт: данные о здоровье чувствительнее токенов.
|
||||
type Collision struct {
|
||||
Metric string
|
||||
Layer string
|
||||
HourUTC time.Time
|
||||
}
|
||||
|
||||
// maxCollisionsReported — сколько координат столкновений попадает в лог.
|
||||
// Больше горсти не нужно: они нужны как зацепка для разбора, а не как отчёт.
|
||||
const maxCollisionsReported = 5
|
||||
|
||||
// MergePoints раскладывает точки по часовым объектам и сливает их с
|
||||
// сохранёнными.
|
||||
//
|
||||
// Час берётся по НАЧАЛУ точки: интервал пересекает границы часов, и любой
|
||||
// другой выбор сделал бы принадлежность объекту зависящей от длительности.
|
||||
// Вся доставка сворачивается ОДНОЙ транзакцией, а не транзакцией на объект.
|
||||
//
|
||||
// Транзакция на объект давала недетерминированное частичное состояние: обход
|
||||
// карты групп рандомизирован, и при отказе посреди доставки набор уже
|
||||
// закоммиченных объектов каждый раз другой (измерено: восемь прогонов одной
|
||||
// доставки — семь разных состояний). Это ломает инвариант «состояние
|
||||
// пересобираемо»: пересборка из архива давала бы не то, что живой приём, а
|
||||
// хеш-детектор после расхождения переписывал бы «неизменившееся».
|
||||
//
|
||||
// Заодно снимается стоимость: отдельный коммит на объект стоил около 0.7 мс,
|
||||
// то есть 8.5 мс на килобайт тела.
|
||||
func (s *Store) MergePoints(ctx context.Context, in []IncomingPoint, deliveryID string) (MergeStats, error) {
|
||||
var stats MergeStats
|
||||
groups := groupByHour(in)
|
||||
keys := sortedKeys(groups)
|
||||
|
||||
for key, group := range groupByHour(in) {
|
||||
res, err := s.mergeBucket(ctx, key, group, deliveryID)
|
||||
if err != nil {
|
||||
return stats, err
|
||||
}
|
||||
stats.Buckets++
|
||||
stats.Points += len(group.points)
|
||||
stats.Overwrites += res.overwrites
|
||||
if res.unchanged {
|
||||
stats.Unchanged++
|
||||
}
|
||||
if res.sealed {
|
||||
stats.SealedHits++
|
||||
var stats MergeStats
|
||||
err := s.inTx(ctx, func(tx *sql.Tx) error {
|
||||
// Счётчики обнуляются на каждой попытке: повтор транзакции начинает
|
||||
// слияние заново, и накопленное от прошлой попытки посчиталось бы дважды.
|
||||
stats = MergeStats{}
|
||||
|
||||
for _, key := range keys {
|
||||
group := groups[key]
|
||||
res, err := mergeBucket(ctx, tx, key, group, deliveryID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
stats.Buckets++
|
||||
stats.Stored += res.stored
|
||||
stats.Overwrites += res.overwrites
|
||||
if res.unchanged {
|
||||
stats.Unchanged++
|
||||
}
|
||||
if res.sealed {
|
||||
stats.SealedHits++
|
||||
}
|
||||
if res.unitsConflict {
|
||||
stats.UnitsConflicts++
|
||||
}
|
||||
if res.overwrites > 0 && len(stats.Collisions) < maxCollisionsReported {
|
||||
stats.Collisions = append(stats.Collisions, Collision{
|
||||
Metric: key.metric, Layer: key.layer, HourUTC: key.hourUTC,
|
||||
})
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return MergeStats{}, err
|
||||
}
|
||||
return stats, nil
|
||||
}
|
||||
|
||||
// sortedKeys задаёт детерминированный порядок обхода объектов доставки.
|
||||
func sortedKeys(groups map[bucketKey]*pointGroup) []bucketKey {
|
||||
keys := make([]bucketKey, 0, len(groups))
|
||||
for k := range groups {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Slice(keys, func(i, j int) bool {
|
||||
if keys[i].metric != keys[j].metric {
|
||||
return keys[i].metric < keys[j].metric
|
||||
}
|
||||
if keys[i].layer != keys[j].layer {
|
||||
return keys[i].layer < keys[j].layer
|
||||
}
|
||||
return keys[i].hourUTC.Before(keys[j].hourUTC)
|
||||
})
|
||||
return keys
|
||||
}
|
||||
|
||||
// IncomingPoint — точка, пришедшая на запись. Отдельный тип от Point потому,
|
||||
// что несёт координаты объекта (метрику, слой, единицы), а Point живёт уже
|
||||
// внутри объекта и их не дублирует.
|
||||
@@ -119,60 +191,62 @@ func groupByHour(in []IncomingPoint) map[bucketKey]*pointGroup {
|
||||
}
|
||||
|
||||
type mergeResult struct {
|
||||
overwrites int
|
||||
unchanged bool
|
||||
sealed bool
|
||||
overwrites int
|
||||
stored int
|
||||
unchanged bool
|
||||
sealed bool
|
||||
unitsConflict bool
|
||||
}
|
||||
|
||||
// mergeBucket выполняет чтение, слияние и запись одного объекта одной
|
||||
// транзакцией.
|
||||
// mergeBucket выполняет чтение, слияние и запись одного объекта.
|
||||
//
|
||||
// Тройка обязана быть атомарной целиком: между чтением и записью может
|
||||
// вклиниться другая доставка того же часа, и её точки пропали бы молча.
|
||||
func (s *Store) mergeBucket(ctx context.Context, key bucketKey, group *pointGroup, deliveryID string) (mergeResult, error) {
|
||||
// Тройка обязана быть атомарной: между чтением и записью может вклиниться
|
||||
// другая доставка того же часа, и её точки пропали бы молча. Атомарность даёт
|
||||
// транзакция вызывающего — она общая на всю доставку.
|
||||
func mergeBucket(ctx context.Context, tx *sql.Tx, key bucketKey, group *pointGroup, deliveryID string) (mergeResult, error) {
|
||||
var res mergeResult
|
||||
|
||||
err := s.inTx(ctx, func(tx *sql.Tx) error {
|
||||
res = mergeResult{}
|
||||
|
||||
stored, found, err := readBucket(ctx, tx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
merged, overwrites := mergePoints(stored.Points, group.points)
|
||||
res.overwrites = overwrites
|
||||
|
||||
hash, err := hashPoints(merged)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if found && hash == stored.Hash {
|
||||
// Хеш — детектор изменений: совпал, значит писать нечего. Именно
|
||||
// это делает широкие проходы синхронизации дешёвыми.
|
||||
res.unchanged = true
|
||||
return nil
|
||||
}
|
||||
res.sealed = found && stored.Sealed
|
||||
|
||||
now := Now()
|
||||
b := Bucket{
|
||||
Metric: key.metric,
|
||||
Layer: key.layer,
|
||||
HourUTC: key.hourUTC,
|
||||
Units: firstNonEmpty(group.units, stored.Units),
|
||||
Points: merged,
|
||||
Hash: hash,
|
||||
Sealed: stored.Sealed,
|
||||
FirstTS: merged[0].Start,
|
||||
LastTS: merged[len(merged)-1].Start,
|
||||
Delivery: firstNonEmpty(stored.Delivery, deliveryID),
|
||||
}
|
||||
return writeBucket(ctx, tx, b, now)
|
||||
})
|
||||
stored, found, err := readBucket(ctx, tx, key)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
|
||||
merged, overwrites := mergePoints(stored.Points, group.points)
|
||||
res.overwrites = overwrites
|
||||
// Считаем сохранённые точки, а не присланные: точные повторы внутри
|
||||
// доставки схлопываются, и счётчик присланных систематически завышал бы
|
||||
// содержимое витрины — расхождение «прислали 1000, лежит 700» было бы
|
||||
// невидимо ровно тогда, когда точки начнут теряться по-настоящему.
|
||||
res.stored = len(merged) - len(stored.Points)
|
||||
res.unitsConflict = stored.Units != "" && group.units != "" && stored.Units != group.units
|
||||
|
||||
hash, err := hashPoints(merged)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
if found && hash == stored.Hash {
|
||||
// Хеш — детектор изменений: совпал, значит писать нечего. Именно это
|
||||
// делает широкие проходы синхронизации дешёвыми.
|
||||
res.unchanged = true
|
||||
return res, nil
|
||||
}
|
||||
res.sealed = found && stored.Sealed
|
||||
|
||||
b := Bucket{
|
||||
Metric: key.metric,
|
||||
Layer: key.layer,
|
||||
HourUTC: key.hourUTC,
|
||||
Units: firstNonEmpty(stored.Units, group.units),
|
||||
Points: merged,
|
||||
Hash: hash,
|
||||
Sealed: stored.Sealed,
|
||||
FirstTS: merged[0].Start,
|
||||
LastTS: merged[len(merged)-1].Start,
|
||||
Delivery: firstNonEmpty(stored.Delivery, deliveryID),
|
||||
}
|
||||
if err := writeBucket(ctx, tx, b, Now()); err != nil {
|
||||
return res, err
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
@@ -196,22 +270,26 @@ func mergePoints(stored, incoming []Point) ([]Point, int) {
|
||||
}
|
||||
|
||||
byCoord := make(map[coord]Point, len(stored)+len(incoming))
|
||||
order := make([]coord, 0, len(stored)+len(incoming))
|
||||
|
||||
put := func(p Point) int {
|
||||
c := coord{start: p.Start.UnixNano(), end: p.End.UnixNano()}
|
||||
old, ok := byCoord[c]
|
||||
if !ok {
|
||||
byCoord[c] = p
|
||||
order = append(order, c)
|
||||
return 0
|
||||
}
|
||||
if bytes.Equal(old.Raw, p.Raw) {
|
||||
// Побайтовое равенство — только быстрый путь. Столкновением считается
|
||||
// расхождение КАНОНИЧЕСКИХ форм: канонизация и заведена потому, что
|
||||
// байты нестабильны. Из 81952 повторно приехавших точек 67534
|
||||
// различаются лишь порядком ключей, ещё 63% — последним разрядом
|
||||
// double. Считай мы по байтам, счётчик перезаписей давал бы тысячи
|
||||
// ложных срабатываний на каждом глубоком проходе, и настоящий отказ
|
||||
// правила слияния стал бы неотличим от нормы.
|
||||
if bytes.Equal(old.Raw, p.Raw) || canon.Equal(old.Raw, p.Raw) {
|
||||
return 0
|
||||
}
|
||||
|
||||
winner := resolve(old, p)
|
||||
byCoord[c] = winner
|
||||
byCoord[c] = resolve(old, p)
|
||||
return 1
|
||||
}
|
||||
|
||||
@@ -223,9 +301,13 @@ func mergePoints(stored, incoming []Point) ([]Point, int) {
|
||||
overwrites += put(p)
|
||||
}
|
||||
|
||||
out := make([]Point, 0, len(order))
|
||||
for _, c := range order {
|
||||
out = append(out, byCoord[c])
|
||||
// Порядок точек в объекте канонический и входит в хеш: его задаёт ТОЛЬКО
|
||||
// сортировка ниже. Порядок обхода карты не специфицирован, и полагаться на
|
||||
// него значило бы получать разные хеши для одного содержимого — тогда
|
||||
// «неизменившийся» объект переписывался бы каждым глубоким проходом.
|
||||
out := make([]Point, 0, len(byCoord))
|
||||
for _, p := range byCoord {
|
||||
out = append(out, p)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool {
|
||||
if !out[i].Start.Equal(out[j].Start) {
|
||||
@@ -378,14 +460,21 @@ func encodePayload(points []Point) ([]byte, error) {
|
||||
})
|
||||
}
|
||||
|
||||
raw, err := json.Marshal(sp)
|
||||
if err != nil {
|
||||
// Через Encoder с выключенным HTML-экранированием, а не json.Marshal:
|
||||
// Marshal превращает `&`, `<` и `>` внутри содержимого точки в
|
||||
// \u0026, \u003c, \u003e. Порчи значений это не даёт, но обещание
|
||||
// «точка хранится дословно» перестаёт быть правдой, а сравнение байтов
|
||||
// при следующей доставке той же точки начинает промахиваться навсегда.
|
||||
var raw bytes.Buffer
|
||||
enc := json.NewEncoder(&raw)
|
||||
enc.SetEscapeHTML(false)
|
||||
if err := enc.Encode(sp); err != nil {
|
||||
return nil, fmt.Errorf("сериализация точек: %w", err)
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
gz := gzip.NewWriter(&buf)
|
||||
if _, err := gz.Write(raw); err != nil {
|
||||
if _, err := gz.Write(raw.Bytes()); err != nil {
|
||||
return nil, fmt.Errorf("сжатие точек: %w", err)
|
||||
}
|
||||
if err := gz.Close(); err != nil {
|
||||
|
||||
@@ -97,12 +97,18 @@ func (s *Store) LastDelivery(ctx context.Context) (Delivery, error) {
|
||||
// Слой сохраняется здесь же, потому что он нужен следующей доставке той же
|
||||
// автоматизации: без плотных метрик выводить его не из чего, и наследовать
|
||||
// приходится от прошлого раза.
|
||||
//
|
||||
// Пустой layer означает «не трогать»: у неудачной свёртки слоя нет, а
|
||||
// затирание оборвало бы цепочку наследования — в том числе у доставки, которая
|
||||
// раньше свернулась успешно, — и результат пересборки журнала изменился бы.
|
||||
func (s *Store) FinishParse(ctx context.Context, id, status string, points int64, layer string) error {
|
||||
const q = `
|
||||
UPDATE delivery SET parse_status = ?, points = ?, derived_layer = ?
|
||||
UPDATE delivery
|
||||
SET parse_status = ?, points = ?,
|
||||
derived_layer = CASE WHEN ? = '' THEN derived_layer ELSE ? END
|
||||
WHERE id = ?`
|
||||
|
||||
res, err := s.db.ExecContext(ctx, q, status, points, layer, id)
|
||||
res, err := s.db.ExecContext(ctx, q, status, points, layer, layer, id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("update parse status: %w", err)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.vakhrushev.me/av/healthlog/internal/store"
|
||||
)
|
||||
|
||||
// Столкновением считается расхождение КАНОНИЧЕСКИХ форм, а не байтов.
|
||||
//
|
||||
// Канонизация и заведена потому, что байты нестабильны: из 81 952 повторно
|
||||
// приехавших точек 67 534 различаются лишь порядком ключей, ещё 63% — последним
|
||||
// разрядом double. Побайтовое сравнение давало тысячи ложных перезаписей на
|
||||
// каждом глубоком проходе, и настоящий отказ правила слияния становился
|
||||
// неотличим от нормы — при том что счётчик перезаписей объявлен единственным
|
||||
// наблюдением за этим правилом.
|
||||
func TestMergePointsДребезгНеСчитаетсяСтолкновением(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := map[string][2]string{
|
||||
"порядок ключей": {
|
||||
`{"date":"a","qty":1.5,"source":"Device A"}`,
|
||||
`{"source":"Device A","qty":1.5,"date":"a"}`,
|
||||
},
|
||||
"последний разряд double": {
|
||||
`{"qty":0.09523182962471353}`,
|
||||
`{"qty":0.09523182962471352}`,
|
||||
},
|
||||
}
|
||||
|
||||
for name, pair := range cases {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
|
||||
first := point(t, "step_count", "minute", "2025-06-05T10:00:00Z", "2025-06-05T10:00:00Z", pair[0])
|
||||
second := point(t, "step_count", "minute", "2025-06-05T10:00:00Z", "2025-06-05T10:00:00Z", pair[1])
|
||||
|
||||
if _, err := st.MergePoints(ctx, []store.IncomingPoint{first}, "d1"); err != nil {
|
||||
t.Fatalf("первое слияние: %v", err)
|
||||
}
|
||||
stats, err := st.MergePoints(ctx, []store.IncomingPoint{second}, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("второе слияние: %v", err)
|
||||
}
|
||||
|
||||
if stats.Overwrites != 0 {
|
||||
t.Errorf("перезаписей %d, ожидалось 0: содержимое то же с точностью до канонической формы",
|
||||
stats.Overwrites)
|
||||
}
|
||||
if len(stats.Collisions) != 0 {
|
||||
t.Errorf("координат столкновений %d, ожидалось 0", len(stats.Collisions))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Настоящее столкновение обязано оставить след с координатами объекта: одно
|
||||
// число `overwrites` не говорит, какая метрика и какой час пострадали, и
|
||||
// расследовать перезапись по нему нечем.
|
||||
func TestMergePointsСтолкновениеОставляетКоординаты(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
|
||||
rich := point(t, "heart_rate", "minute", "2025-06-05T10:00:00Z", "2025-06-05T10:00:00Z",
|
||||
`{"Avg":60,"context":"Отдых"}`)
|
||||
poor := point(t, "heart_rate", "minute", "2025-06-05T10:00:00Z", "2025-06-05T10:00:00Z",
|
||||
`{"Avg":61}`)
|
||||
|
||||
if _, err := st.MergePoints(ctx, []store.IncomingPoint{rich}, "d1"); err != nil {
|
||||
t.Fatalf("первое слияние: %v", err)
|
||||
}
|
||||
stats, err := st.MergePoints(ctx, []store.IncomingPoint{poor}, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("второе слияние: %v", err)
|
||||
}
|
||||
|
||||
if len(stats.Collisions) != 1 {
|
||||
t.Fatalf("координат столкновений %d, ожидалась 1", len(stats.Collisions))
|
||||
}
|
||||
c := stats.Collisions[0]
|
||||
if c.Metric != "heart_rate" || c.Layer != "minute" {
|
||||
t.Errorf("координаты столкновения %s/%s, ожидались heart_rate/minute", c.Metric, c.Layer)
|
||||
}
|
||||
if c.HourUTC.Hour() != 10 {
|
||||
t.Errorf("час столкновения %v", c.HourUTC)
|
||||
}
|
||||
}
|
||||
|
||||
// «Точка хранится дословно» читается буквально: байты как пришли. json.Marshal
|
||||
// экранирует `&`, `<` и `>` внутри содержимого — порчи значений это не даёт, но
|
||||
// обещание перестаёт быть правдой, а сравнение байтов при следующей доставке
|
||||
// той же точки начинает промахиваться навсегда.
|
||||
func TestMergePointsХранитУгловыеСкобкиИАмперсанд(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
|
||||
raw := `{"qty":1,"source":"Anton & Co <iPhone> \"x\""}`
|
||||
in := point(t, "step_count", "minute", "2025-06-05T10:00:00Z", "2025-06-05T10:00:00Z", raw)
|
||||
|
||||
if _, err := st.MergePoints(ctx, []store.IncomingPoint{in}, "d1"); err != nil {
|
||||
t.Fatalf("слияние: %v", err)
|
||||
}
|
||||
|
||||
b, err := st.Bucket(ctx, "step_count", "minute", ts(t, "2025-06-05T10:00:00Z"))
|
||||
if err != nil {
|
||||
t.Fatalf("чтение: %v", err)
|
||||
}
|
||||
got := string(b.Points[0].Raw)
|
||||
if got != raw {
|
||||
t.Errorf("точка изменилась при хранении:\n было %s\n стало %s", raw, got)
|
||||
}
|
||||
for _, escaped := range []string{"\\u0026", "\\u003c", "\\u003e"} {
|
||||
if strings.Contains(got, escaped) {
|
||||
t.Errorf("содержимое точки экранировано как HTML (%s): %s", escaped, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Смена единиц не имеет права молча переподписать уже сохранённые точки:
|
||||
// внутри точки единиц нет, и у ранних точек не остаётся ничего, по чему их
|
||||
// единицы восстановимы. Правило «первое непустое побеждает» плюс счётчик.
|
||||
func TestMergePointsСменаЕдиницНеПерезаписываетМолча(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
|
||||
km := store.IncomingPoint{
|
||||
Metric: "walking_running_distance", Layer: "minute", Units: "km",
|
||||
Point: store.Point{
|
||||
Start: ts(t, "2025-06-05T10:00:00Z"), End: ts(t, "2025-06-05T10:00:00Z"),
|
||||
Raw: []byte(`{"qty":1}`),
|
||||
},
|
||||
}
|
||||
mi := km
|
||||
mi.Units = "mi"
|
||||
mi.Start = ts(t, "2025-06-05T10:30:00Z")
|
||||
mi.End = mi.Start
|
||||
mi.Raw = []byte(`{"qty":2}`)
|
||||
|
||||
if _, err := st.MergePoints(ctx, []store.IncomingPoint{km}, "d1"); err != nil {
|
||||
t.Fatalf("первое слияние: %v", err)
|
||||
}
|
||||
stats, err := st.MergePoints(ctx, []store.IncomingPoint{mi}, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("второе слияние: %v", err)
|
||||
}
|
||||
|
||||
if stats.UnitsConflicts != 1 {
|
||||
t.Errorf("расхождений единиц %d, ожидалось 1", stats.UnitsConflicts)
|
||||
}
|
||||
|
||||
b, err := st.Bucket(ctx, "walking_running_distance", "minute", ts(t, "2025-06-05T10:00:00Z"))
|
||||
if err != nil {
|
||||
t.Fatalf("чтение: %v", err)
|
||||
}
|
||||
if b.Units != "km" {
|
||||
t.Errorf("единицы объекта %q: сохранённые точки переподписаны чужими", b.Units)
|
||||
}
|
||||
}
|
||||
|
||||
// Счётчик считает СОХРАНЁННЫЕ точки, а не присланные: точные повторы внутри
|
||||
// доставки схлопываются, и счётчик присланных систематически завышал бы
|
||||
// содержимое витрины — расхождение «прислали 1000, лежит 700» было бы невидимо
|
||||
// ровно тогда, когда точки начнут теряться по-настоящему.
|
||||
func TestMergePointsСчитаетСохранённые(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
|
||||
p := point(t, "step_count", "minute", "2025-06-05T10:00:00Z", "2025-06-05T10:00:00Z", `{"qty":1}`)
|
||||
stats, err := st.MergePoints(ctx, []store.IncomingPoint{p, p, p}, "d1")
|
||||
if err != nil {
|
||||
t.Fatalf("слияние: %v", err)
|
||||
}
|
||||
if stats.Stored != 1 {
|
||||
t.Errorf("сохранено %d точек, ожидалась 1 (три точных повтора)", stats.Stored)
|
||||
}
|
||||
|
||||
stats, err = st.MergePoints(ctx, []store.IncomingPoint{p}, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("повтор: %v", err)
|
||||
}
|
||||
if stats.Stored != 0 {
|
||||
t.Errorf("повтор добавил %d точек, ожидалось 0", stats.Stored)
|
||||
}
|
||||
}
|
||||
|
||||
// Доставка сворачивается одной транзакцией, поэтому отказ посреди неё не
|
||||
// оставляет частичного состояния — а значит и не зависит от порядка обхода.
|
||||
// Раньше транзакция была на объект, и восемь прогонов одной доставки давали
|
||||
// семь разных наборов записанных объектов.
|
||||
func TestMergePointsОтказНеОставляетЧастиОбъектов(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
|
||||
in := make([]store.IncomingPoint, 0, 40)
|
||||
for i := range 40 {
|
||||
at := ts(t, "2025-06-05T00:00:00Z").Add(time.Duration(i) * time.Hour)
|
||||
in = append(in, store.IncomingPoint{
|
||||
Metric: "step_count", Layer: "minute", Units: "count",
|
||||
Point: store.Point{Start: at, End: at, Raw: []byte(`{"qty":1}`)},
|
||||
})
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
if _, err := st.MergePoints(ctx, in, "d1"); err == nil {
|
||||
t.Fatal("слияние на отменённом контексте прошло успешно")
|
||||
}
|
||||
|
||||
n, err := st.CountBuckets(context.Background())
|
||||
if err != nil {
|
||||
t.Fatalf("счёт объектов: %v", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Errorf("записано %d объектов из 40: отказ оставил частичное состояние", n)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user