первая встреча непокрытой секции стала наблюдаемым событием
- свёртка спрашивает журнал, встречалось ли имя строго раньше по паре (received_at, id), и пишет WARN с атрибутом uncovered_new; повторные молчат. Признак выводится, а не хранится — реестр был бы второй копией факта - добавлена подкоманда `healthlog uncovered`: перечень накопленного, чтение только на чтение, экранированные имена и названные границы носителя - синк документации: ADR о выводе новизны из журнала, две записи в журнал дефектов, два правила промоутом в конвенции, терминал оператора назван адресатом недоверенного входа
This commit is contained in:
+119
-4
@@ -92,6 +92,17 @@ type Stats struct {
|
||||
Uncovered []string
|
||||
// UncoveredDropped — сколько имён отброшено границей списка.
|
||||
UncoveredDropped int
|
||||
// UncoveredNew — имена непокрытых секций, которых не было ни в одной
|
||||
// доставке, стоящей в журнале раньше этой. Событие однократное за всю жизнь
|
||||
// имени: поток дописывает метрики на телефоне молча, и момент появления
|
||||
// секции наблюдать больше нечем.
|
||||
UncoveredNew []string
|
||||
// UncoveredSeenUnknown — сверка с журналом не состоялась, и потому все
|
||||
// непокрытые имена доставки объявлены новыми. Лишняя запись стоит внимания
|
||||
// один раз, промолчавшее событие не восстанавливается ничем.
|
||||
UncoveredSeenUnknown bool
|
||||
// UncoveredSeenError — почему не состоялась.
|
||||
UncoveredSeenError error
|
||||
|
||||
// Categoricals — сколько РАЗЛИЧНЫХ категориальных значений наблюдалось;
|
||||
// CategoricalUnknown — сколько из них словарь не знает;
|
||||
@@ -158,6 +169,16 @@ func (s *Service) Fold(ctx context.Context, deliveryID string) (stats Stats, err
|
||||
FallbackLayer: hae.Layer(fallback),
|
||||
Locale: localeOf(d.Headers),
|
||||
})
|
||||
|
||||
// Сверка с журналом идёт ДО ветвления на успех и отказ. Список непокрытых
|
||||
// секций переживает отказ разбора, то есть имя уже записано в учёт; смолчи
|
||||
// здесь — и следующая доставка сочтёт его виденным, а событие не вернётся
|
||||
// ничем, кроме ручного запроса в базу.
|
||||
novelty := s.novelty(ctx, parsed.Uncovered, store.DeliveryRef{
|
||||
ID: d.ID,
|
||||
ReceivedAt: d.ReceivedAt,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
// Список непокрытых секций переживает отказ: доставка, у которой не
|
||||
// определился слой, обязана остаться записью о том, что в теле есть
|
||||
@@ -166,12 +187,15 @@ func (s *Service) Fold(ctx context.Context, deliveryID string) (stats Stats, err
|
||||
// А вот число пропущенных сущностей — НЕ переживает: разбор, вернувший
|
||||
// ошибку, отдаёт нулевые счётчики по построению, а не по измерению, и
|
||||
// записать этот ноль значило бы объявить доставку проверенной.
|
||||
s.fail(ctx, deliveryID, err, residueOf(parsed))
|
||||
s.fail(ctx, deliveryID, err, residueOf(parsed, novelty))
|
||||
return stats, err
|
||||
}
|
||||
|
||||
stats.Uncovered = parsed.Uncovered
|
||||
stats.UncoveredDropped = parsed.UncoveredDropped
|
||||
stats.UncoveredNew = novelty.fresh
|
||||
stats.UncoveredSeenUnknown = novelty.unknown
|
||||
stats.UncoveredSeenError = novelty.cause
|
||||
stats.Metrics = parsed.Metrics
|
||||
stats.Points = len(parsed.Points)
|
||||
stats.SkippedNoTime = parsed.SkippedNoTime
|
||||
@@ -196,6 +220,11 @@ func (s *Service) Fold(ctx context.Context, deliveryID string) (stats Stats, err
|
||||
s.fail(ctx, deliveryID, err, parseResidue{
|
||||
uncovered: parsed.Uncovered,
|
||||
skipped: skippedEntities(parsed),
|
||||
// Новизна доезжает и сюда. Эта ветвь пишет имя в учёт ровно так же,
|
||||
// как ветвь отказа разбора, — значит и терять событие ей нельзя:
|
||||
// следующая доставка сочтёт имя виденным, а отказ слияния бывает
|
||||
// нетранзиентным (исчерпанный дедлайн свёртки под большим телом).
|
||||
novelty: novelty,
|
||||
})
|
||||
return stats, err
|
||||
}
|
||||
@@ -283,6 +312,18 @@ func (s *Service) logResult(ctx context.Context, deliveryID string, st Stats) {
|
||||
// построчный разбор логов. Содержимого секций здесь нет.
|
||||
"uncovered", st.Uncovered,
|
||||
"uncovered_dropped", st.UncoveredDropped,
|
||||
// Имена, встреченные впервые по журналу, — атрибутом ВСЕГДА, а уровень
|
||||
// поднимается отдельной ветвью ниже. Наблюдаемый признак события это
|
||||
// он: имя непокрытой секции стоит в атрибуте `uncovered` у каждой
|
||||
// доставки, которая её принесла, и по нему первую встречу не отличить.
|
||||
"uncovered_new", st.UncoveredNew,
|
||||
"uncovered_seen_unknown", st.UncoveredSeenUnknown,
|
||||
// Причина несостоявшейся сверки — рядом с признаком. Занятость базы
|
||||
// проходит сама, испорченная колонка не проходит никогда и поднимает
|
||||
// признак на каждой доставке; по одному булеву это неразличимо.
|
||||
// Значений точек в ошибке нет: до текста доезжает только имя секции, и
|
||||
// оно обрезано.
|
||||
"uncovered_seen_error", st.UncoveredSeenError,
|
||||
// Категориальные значения — ЧИСЛАМИ. Ни строк, ни выведенных кодов:
|
||||
// «Сидячий образ жизни» — это контекст пульса, то есть данные о
|
||||
// здоровье. Какие именно строки ждут словаря, отвечает реестр в базе.
|
||||
@@ -318,6 +359,15 @@ func (s *Service) logResult(ctx context.Context, deliveryID string, st Stats) {
|
||||
allEntitiesSkipped := st.Workouts == 0 && st.Records == 0 && skippedEntities > 0
|
||||
|
||||
switch {
|
||||
case len(st.UncoveredNew) > 0:
|
||||
// ПЕРВОЙ ветвью, и это существенно. Все прочие говорят о событиях,
|
||||
// повторяющихся на живом потоке; это — однократное за всю жизнь имени, и
|
||||
// замаскировать его перезаписью точек значило бы потерять ровно то, ради
|
||||
// чего наблюдение заведено. Уровень `WARN`, а не `ERROR`: приезд новой
|
||||
// секции — штатное событие внешнего мира, «посмотри», а не «разбери
|
||||
// сбой». Имена секций в лог попадать могут: имя ключа — форма пакета, а
|
||||
// не измерение.
|
||||
s.log.WarnContext(ctx, "delivery folded, new uncovered section", attrs...)
|
||||
case st.EntitiesHeld > 0:
|
||||
// Приехавшая версия сущности отклонена как теряющая содержание. Плата
|
||||
// за отказ объединять поля: событие обязано быть видно, потому что на
|
||||
@@ -446,10 +496,61 @@ type parseResidue struct {
|
||||
// считал». Ноль означал бы «проверено, терять нечего», а по этому числу
|
||||
// ретеншен принимает необратимое решение об удалении тела.
|
||||
skipped *int64
|
||||
// novelty в учёте не участвует — она едет в запись лога об отказе. Полем, а
|
||||
// не пятым параметром `fail`: параметры путают местами, а поле называет
|
||||
// себя само.
|
||||
novelty sectionNovelty
|
||||
}
|
||||
|
||||
func residueOf(parsed hae.Result) parseResidue {
|
||||
return parseResidue{uncovered: parsed.Uncovered}
|
||||
func residueOf(parsed hae.Result, novelty sectionNovelty) parseResidue {
|
||||
return parseResidue{uncovered: parsed.Uncovered, novelty: novelty}
|
||||
}
|
||||
|
||||
// sectionNovelty — исход сверки имён непокрытых секций с журналом.
|
||||
type sectionNovelty struct {
|
||||
// fresh — имена, которых не было ни в одной доставке раньше этой.
|
||||
fresh []string
|
||||
// unknown — сверка не состоялась, и потому новыми объявлены ВСЕ имена
|
||||
// доставки.
|
||||
unknown bool
|
||||
// cause — почему не состоялась. Без неё занятость базы (пройдёт сама) и
|
||||
// испорченное содержимое колонки (не пройдёт никогда, и признак будет
|
||||
// подниматься на каждой доставке) неотличимы, а разбираться пришлось бы тем
|
||||
// самым ручным запросом в базу, от которого задача избавляет.
|
||||
cause error
|
||||
}
|
||||
|
||||
// novelty спрашивает журнал, какие из непокрытых имён встречаются впервые.
|
||||
//
|
||||
// Отказ запроса свёртку не роняет и исходом доставки не становится: правила
|
||||
// классификации исходов наблюдение не трогает, занятая база и отменённый
|
||||
// контекст остаются обстоятельствами. Но и молчания здесь быть не может — имя,
|
||||
// о котором смолчали, уже записано в учёт, — поэтому при отказе новыми
|
||||
// объявляются все имена, а признак несостоявшейся сверки идёт в запись.
|
||||
//
|
||||
// Запрос берётся только при непустом списке: на живом потоке все три
|
||||
// приезжающие секции покрыты, то есть в штатном режиме сверка не стоит ничего.
|
||||
func (s *Service) novelty(ctx context.Context, uncovered []string, at store.DeliveryRef) sectionNovelty {
|
||||
if len(uncovered) == 0 {
|
||||
return sectionNovelty{}
|
||||
}
|
||||
|
||||
seen, err := s.store.SectionsSeenBefore(ctx, uncovered, at)
|
||||
if err != nil {
|
||||
return sectionNovelty{fresh: uncovered, unknown: true, cause: err}
|
||||
}
|
||||
|
||||
fresh := make([]string, 0, len(uncovered))
|
||||
for _, name := range uncovered {
|
||||
if _, ok := seen[name]; ok {
|
||||
continue
|
||||
}
|
||||
fresh = append(fresh, name)
|
||||
}
|
||||
if len(fresh) == 0 {
|
||||
return sectionNovelty{}
|
||||
}
|
||||
return sectionNovelty{fresh: fresh}
|
||||
}
|
||||
|
||||
// skippedEntities — сколько сущностей с собственным `id` разбор пропустил.
|
||||
@@ -469,6 +570,20 @@ func (s *Service) fail(ctx context.Context, deliveryID string, cause error, resi
|
||||
return
|
||||
}
|
||||
|
||||
// Новые имена доезжают до записи об отказе: она уже выше рутинного уровня,
|
||||
// а событие опознаётся атрибутом. В отложенном исходе выше их нет намеренно
|
||||
// — там учётная запись не меняется, доставка вернётся следующим проходом, и
|
||||
// повторение признака на каждом проходе занятой базы превратило бы
|
||||
// однократное событие в дребезг.
|
||||
attrs := []any{"error", cause, "delivery_id", deliveryID}
|
||||
if len(residue.novelty.fresh) > 0 {
|
||||
attrs = append(attrs, "uncovered_new", residue.novelty.fresh)
|
||||
}
|
||||
if residue.novelty.unknown {
|
||||
attrs = append(attrs, "uncovered_seen_unknown", true,
|
||||
"uncovered_seen_error", residue.novelty.cause)
|
||||
}
|
||||
|
||||
level := slog.LevelError
|
||||
switch {
|
||||
case errors.Is(cause, hae.ErrLayerUnknown):
|
||||
@@ -481,7 +596,7 @@ func (s *Service) fail(ctx context.Context, deliveryID string, cause error, resi
|
||||
// одного класса ошибки давали бы постоянный ERROR-шум.
|
||||
level = slog.LevelWarn
|
||||
}
|
||||
s.log.Log(ctx, level, "delivery fold failed", "error", cause, "delivery_id", deliveryID)
|
||||
s.log.Log(ctx, level, "delivery fold failed", attrs...)
|
||||
|
||||
// Слой НЕ затирается: доставка могла свернуться успешно раньше, и пустая
|
||||
// строка здесь оборвала бы цепочку наследования, то есть изменила бы
|
||||
|
||||
@@ -238,6 +238,15 @@ func TestFoldЧастичныйРазборВЛоге(t *testing.T) {
|
||||
t.Fatalf("свёртка: %v", err)
|
||||
}
|
||||
|
||||
// Смотрим на ВТОРУЮ доставку с той же секцией. Первая встреча имени —
|
||||
// событие само по себе и уровень поднимает законно; здесь же проверяется
|
||||
// другое: что уровень не поднимает сама по себе частичность, установившееся
|
||||
// состояние половины потока.
|
||||
deliver(t, arch, st, "d2", "Minutes", "a1", []byte(body))
|
||||
if _, err := f.Fold(ctx, "d2"); err != nil {
|
||||
t.Fatalf("повторная свёртка: %v", err)
|
||||
}
|
||||
|
||||
out := buf.String()
|
||||
if !strings.Contains(out, "ecg") {
|
||||
t.Error("имени непокрытой секции нет в логе — момент появления новой секции незаметен")
|
||||
|
||||
@@ -0,0 +1,138 @@
|
||||
package fold
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.vakhrushev.me/av/healthlog/internal/store"
|
||||
)
|
||||
|
||||
// Внутренний тест, а не через `Fold`, и это осознанно. Сломать сверку снаружи
|
||||
// больше нечем: испорченную учётную запись запрос теперь пропускает, а
|
||||
// единственная настоящая причина отказа — отмена контекста и занятость базы —
|
||||
// в полном проходе свёртки утащила бы за собой и слияние, то есть проверялась
|
||||
// бы уже другая ветвь. Заводить интерфейс хранилища ради мока дороже: он
|
||||
// пережил бы тест и остался бы в коде навсегда.
|
||||
func TestNoveltyОтменённыйКонтекстНеМолчит(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
st, err := store.Open(filepath.Join(dir, "healthlog.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("база: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = st.Close() })
|
||||
|
||||
s := &Service{store: st, log: slog.New(slog.DiscardHandler)}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
got := s.novelty(ctx, []string{"ecg", "symptoms"}, store.DeliveryRef{
|
||||
ID: "d1",
|
||||
ReceivedAt: time.Now().UTC(),
|
||||
})
|
||||
|
||||
if !got.unknown {
|
||||
t.Fatal("сверка на отменённом контексте объявила себя состоявшейся")
|
||||
}
|
||||
if len(got.fresh) != 2 {
|
||||
t.Errorf("новыми объявлены %v, ожидались все имена доставки", got.fresh)
|
||||
}
|
||||
if got.cause == nil {
|
||||
t.Error("причина отказа потеряна: занятость базы и вечную порчу колонки по признаку не различить")
|
||||
}
|
||||
}
|
||||
|
||||
// Пустой список сверку не берёт вовсе: на живом потоке все три приезжающие
|
||||
// секции покрыты, и запрос по журналу в штатном режиме не стоит ничего.
|
||||
func TestNoveltyПустойСписокЗапросаНеБерёт(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
s := &Service{store: nil, log: slog.New(slog.DiscardHandler)}
|
||||
|
||||
// Хранилище nil: возьмись запрос — тест упал бы паникой.
|
||||
got := s.novelty(context.Background(), nil, store.DeliveryRef{ID: "d1"})
|
||||
if got.unknown || len(got.fresh) != 0 {
|
||||
t.Errorf("на пустом списке сверка что-то решила: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Признак несостоявшейся сверки без причины неразбираем: занятость базы пройдёт
|
||||
// сама, испорченная колонка не пройдёт никогда и будет поднимать признак на
|
||||
// каждой доставке.
|
||||
func TestLogResultПечатаетПричинуНесостоявшейсяСверки(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var buf bytes.Buffer
|
||||
s := &Service{log: slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo}))}
|
||||
|
||||
s.logResult(context.Background(), "d1", Stats{
|
||||
Uncovered: []string{"ecg"},
|
||||
UncoveredNew: []string{"ecg"},
|
||||
UncoveredSeenUnknown: true,
|
||||
UncoveredSeenError: errors.New("журнал недоступен"),
|
||||
})
|
||||
|
||||
var rec map[string]any
|
||||
line := strings.TrimSpace(buf.String())
|
||||
if err := json.Unmarshal([]byte(line), &rec); err != nil {
|
||||
t.Fatalf("запись не разбирается: %v", err)
|
||||
}
|
||||
if rec["uncovered_seen_unknown"] != true {
|
||||
t.Error("признак несостоявшейся сверки не выставлен")
|
||||
}
|
||||
if rec["uncovered_seen_error"] != "журнал недоступен" {
|
||||
t.Errorf("причина в записи: %v", rec["uncovered_seen_error"])
|
||||
}
|
||||
if rec["level"] != "WARN" {
|
||||
t.Errorf("уровень %v, ожидался WARN: имена объявлены новыми", rec["level"])
|
||||
}
|
||||
}
|
||||
|
||||
// Запись об отказе несёт и признак несостоявшейся сверки, и причину: у отказа
|
||||
// разбора та же цена молчания, что у успешного пути — имя уже в учёте.
|
||||
func TestFailПечатаетНесостоявшуюсяСверку(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
st, err := store.Open(filepath.Join(dir, "healthlog.db"))
|
||||
if err != nil {
|
||||
t.Fatalf("база: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = st.Close() })
|
||||
|
||||
ctx := context.Background()
|
||||
err = st.CreateDelivery(ctx, store.Delivery{
|
||||
ID: "d1", ReceivedAt: time.Now().UTC(), RawPath: "d1.json.gz",
|
||||
SHA256: "-", ParseStatus: store.ParsePending,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("учёт доставки: %v", err)
|
||||
}
|
||||
|
||||
var buf bytes.Buffer
|
||||
s := &Service{
|
||||
store: st,
|
||||
log: slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})),
|
||||
}
|
||||
|
||||
s.fail(ctx, "d1", errors.New("непонятое содержимое"), parseResidue{
|
||||
uncovered: []string{"ecg"},
|
||||
novelty: sectionNovelty{fresh: []string{"ecg"}, unknown: true, cause: errors.New("журнал недоступен")},
|
||||
})
|
||||
|
||||
out := buf.String()
|
||||
for _, want := range []string{"uncovered_new", "ecg", "uncovered_seen_unknown", "журнал недоступен"} {
|
||||
if !strings.Contains(out, want) {
|
||||
t.Errorf("в записи об отказе нет %q:\n%s", want, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,395 @@
|
||||
package fold_test
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.vakhrushev.me/av/healthlog/internal/archive"
|
||||
"git.vakhrushev.me/av/healthlog/internal/fold"
|
||||
"git.vakhrushev.me/av/healthlog/internal/store"
|
||||
|
||||
_ "modernc.org/sqlite" // прямое подключение к файлу базы — ради теста, ломающего сверку
|
||||
)
|
||||
|
||||
// loggedFold — свёртка с логгером, чьи записи можно прочитать.
|
||||
//
|
||||
// Отдельный конструктор, потому что событие о новой секции наблюдаемо ТОЛЬКО
|
||||
// через лог: в учёте его нет и быть не может — колонка говорит, какие секции в
|
||||
// теле есть, а не какая из них встретилась впервые. Путь к файлу базы нужен
|
||||
// тесту, ломающему сверку: сломать её изнутри нечем — свёртка держит настоящее
|
||||
// хранилище, а заводить интерфейс ради мока значит менять код под тест.
|
||||
type loggedFold struct {
|
||||
svc *fold.Service
|
||||
arch *archive.Archive
|
||||
st *store.Store
|
||||
log *bytes.Buffer
|
||||
dbPath string
|
||||
}
|
||||
|
||||
func newLoggedFold(t *testing.T) loggedFold {
|
||||
t.Helper()
|
||||
|
||||
dir := t.TempDir()
|
||||
arch, err := archive.New(filepath.Join(dir, "raw"))
|
||||
if err != nil {
|
||||
t.Fatalf("архив: %v", err)
|
||||
}
|
||||
dbPath := filepath.Join(dir, "healthlog.db")
|
||||
st, err := store.Open(dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("база: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = st.Close() })
|
||||
|
||||
var buf bytes.Buffer
|
||||
log := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
||||
return loggedFold{
|
||||
svc: fold.New(arch, st, 0, log),
|
||||
arch: arch,
|
||||
st: st,
|
||||
log: &buf,
|
||||
dbPath: dbPath,
|
||||
}
|
||||
}
|
||||
|
||||
// breakMerge сносит таблицу объектов: слияние отказывает нетранзиентно, а разбор
|
||||
// и запись исхода продолжают работать.
|
||||
//
|
||||
// В жизни в ту же ветвь ведёт исчерпанный дедлайн свёртки под большим телом
|
||||
// (измерено: 63 МиБ держат блокировку 5.019 с) — `store.Transient` дедлайн
|
||||
// намеренно не признаёт.
|
||||
func breakMerge(t *testing.T, dbPath string) {
|
||||
t.Helper()
|
||||
|
||||
db, err := sql.Open("sqlite", "file:"+dbPath+"?_pragma=busy_timeout(5000)")
|
||||
if err != nil {
|
||||
t.Fatalf("открытие базы напрямую: %v", err)
|
||||
}
|
||||
defer func() { _ = db.Close() }()
|
||||
|
||||
if _, err := db.ExecContext(context.Background(), `DROP TABLE bucket`); err != nil {
|
||||
t.Fatalf("снос таблицы объектов: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// logRecords разбирает захваченные записи лога.
|
||||
func logRecords(t *testing.T, buf *bytes.Buffer) []map[string]any {
|
||||
t.Helper()
|
||||
|
||||
var out []map[string]any
|
||||
for line := range strings.SplitSeq(strings.TrimSpace(buf.String()), "\n") {
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var rec map[string]any
|
||||
if err := json.Unmarshal([]byte(line), &rec); err != nil {
|
||||
t.Fatalf("строка лога не JSON: %v", err)
|
||||
}
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// newSections достаёт имена новых секций из записи лога.
|
||||
func newSections(t *testing.T, rec map[string]any) []string {
|
||||
t.Helper()
|
||||
|
||||
raw, ok := rec["uncovered_new"]
|
||||
if !ok || raw == nil {
|
||||
// Пустой список кодировщик пишет как `null` — это и есть «новых имён
|
||||
// нет», а не сломанный атрибут.
|
||||
return nil
|
||||
}
|
||||
list, ok := raw.([]any)
|
||||
if !ok {
|
||||
t.Fatalf("атрибут uncovered_new не список: %#v", raw)
|
||||
}
|
||||
out := make([]string, 0, len(list))
|
||||
for _, v := range list {
|
||||
s, ok := v.(string)
|
||||
if !ok {
|
||||
t.Fatalf("имя секции не строка: %#v", v)
|
||||
}
|
||||
out = append(out, s)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// withSection дописывает в тело непокрытую секцию с данным именем.
|
||||
func withSection(t *testing.T, body []byte, name string) []byte {
|
||||
t.Helper()
|
||||
|
||||
const anchor = `"data": {`
|
||||
if !bytes.Contains(body, []byte(anchor)) {
|
||||
t.Fatalf("в теле нет объекта data — дописать секцию некуда")
|
||||
}
|
||||
return bytes.Replace(body, []byte(anchor), []byte(anchor+"\n \""+name+"\": [],"), 1)
|
||||
}
|
||||
|
||||
// Момент появления секции наблюдать больше нечем: телефон дописывает метрики
|
||||
// молча, а колонка учёта говорит, какие секции в теле есть, а не какая из них
|
||||
// приехала впервые.
|
||||
func TestFoldПерваяВстречаСекцииДаётСобытие(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
body := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||||
|
||||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", body)
|
||||
stats, err := lf.svc.Fold(ctx, "d1")
|
||||
if err != nil {
|
||||
t.Fatalf("свёртка: %v", err)
|
||||
}
|
||||
|
||||
if len(stats.UncoveredNew) != 1 || stats.UncoveredNew[0] != "symptoms" {
|
||||
t.Fatalf("новых секций %v, ожидалась ровно `symptoms`", stats.UncoveredNew)
|
||||
}
|
||||
|
||||
recs := logRecords(t, lf.log)
|
||||
var announced int
|
||||
for _, rec := range recs {
|
||||
for _, name := range newSections(t, rec) {
|
||||
if name != "symptoms" {
|
||||
continue
|
||||
}
|
||||
announced++
|
||||
if rec["level"] != "WARN" {
|
||||
t.Errorf("уровень записи %v, ожидался WARN: приезд новой секции — «посмотри», а не «разбери сбой»",
|
||||
rec["level"])
|
||||
}
|
||||
}
|
||||
}
|
||||
if announced != 1 {
|
||||
t.Errorf("записей о новой секции %d, ожидалась ровно одна", announced)
|
||||
}
|
||||
}
|
||||
|
||||
// Вторая доставка с тем же именем молчит: событие однократно за всю жизнь
|
||||
// имени, иначе поток раз в пять минут обесценил бы уровень.
|
||||
func TestFoldВтораяВстречаСобытияНеДаёт(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
body := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||||
|
||||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", body)
|
||||
if _, err := lf.svc.Fold(ctx, "d1"); err != nil {
|
||||
t.Fatalf("первая свёртка: %v", err)
|
||||
}
|
||||
lf.log.Reset()
|
||||
|
||||
deliver(t, lf.arch, lf.st, "d2", "Minutes", "auto-1", body)
|
||||
stats, err := lf.svc.Fold(ctx, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("вторая свёртка: %v", err)
|
||||
}
|
||||
|
||||
if len(stats.UncoveredNew) != 0 {
|
||||
t.Errorf("вторая доставка объявила новыми %v", stats.UncoveredNew)
|
||||
}
|
||||
|
||||
recs := logRecords(t, lf.log)
|
||||
if len(recs) == 0 {
|
||||
t.Fatal("вторая свёртка не записала ничего — чекпоинт молчит вовсе")
|
||||
}
|
||||
for _, rec := range recs {
|
||||
if len(newSections(t, rec)) != 0 {
|
||||
t.Errorf("вторая доставка назвала новые секции: %v", rec)
|
||||
}
|
||||
if rec["msg"] == "delivery folded" && rec["level"] != "INFO" {
|
||||
t.Errorf("уровень повторной доставки %v, ожидался INFO", rec["level"])
|
||||
}
|
||||
}
|
||||
|
||||
// Перечисление непокрытых секций при этом не меняется: атрибут `uncovered`
|
||||
// продолжает называть их все — по нему первую встречу и не отличить.
|
||||
if !strings.Contains(lf.log.String(), "symptoms") {
|
||||
t.Error("имя секции пропало из записи вовсе: атрибут непокрытых секций не трогается")
|
||||
}
|
||||
}
|
||||
|
||||
// Новое имя рядом с уже виденным: событие адресное, а не «в теле что-то новое».
|
||||
func TestFoldНовоеИмяРядомСВиденным(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
first := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||||
second := withSection(t, first, "ecg")
|
||||
|
||||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", first)
|
||||
if _, err := lf.svc.Fold(ctx, "d1"); err != nil {
|
||||
t.Fatalf("первая свёртка: %v", err)
|
||||
}
|
||||
|
||||
deliver(t, lf.arch, lf.st, "d2", "Minutes", "auto-1", second)
|
||||
stats, err := lf.svc.Fold(ctx, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("вторая свёртка: %v", err)
|
||||
}
|
||||
|
||||
if len(stats.UncoveredNew) != 1 || stats.UncoveredNew[0] != "ecg" {
|
||||
t.Errorf("новых секций %v, ожидалась ровно `ecg`", stats.UncoveredNew)
|
||||
}
|
||||
}
|
||||
|
||||
// Событие однократно за жизнь имени, значит замаскировать его рутинным
|
||||
// событием нельзя: перезаписи точек случаются на живом потоке постоянно, а
|
||||
// приезд новой секции — один раз.
|
||||
func TestFoldНоваяСекцияНеМаскируетсяПерезаписью(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
|
||||
base := fixture(t, "minute.json")
|
||||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", base)
|
||||
if _, err := lf.svc.Fold(ctx, "d1"); err != nil {
|
||||
t.Fatalf("первая свёртка: %v", err)
|
||||
}
|
||||
lf.log.Reset()
|
||||
|
||||
// Те же координаты, другое значение: набор полей совпадает, полнота равна,
|
||||
// поэтому побеждает пришедшая — это перезапись.
|
||||
changed := regexp.MustCompile(`"qty"\s*:\s*[-0-9.eE+]+`).ReplaceAll(base, []byte(`"qty": 777.5`))
|
||||
if bytes.Equal(changed, base) {
|
||||
t.Fatal("фикстура не содержит qty — перезапись не устроить")
|
||||
}
|
||||
deliver(t, lf.arch, lf.st, "d2", "Minutes", "auto-1", withSection(t, changed, "ecg"))
|
||||
stats, err := lf.svc.Fold(ctx, "d2")
|
||||
if err != nil {
|
||||
t.Fatalf("вторая свёртка: %v", err)
|
||||
}
|
||||
if stats.Overwrites == 0 {
|
||||
t.Fatal("перезаписей нет — рутинного события, которое могло бы замаскировать новое, не случилось")
|
||||
}
|
||||
|
||||
var found bool
|
||||
for _, rec := range logRecords(t, lf.log) {
|
||||
if rec["msg"] != "delivery folded, new uncovered section" {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if rec["overwrites"] == nil {
|
||||
t.Error("счётчик перезаписей исчез из записи: признаки идут атрибутами всегда")
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("запись назвала перезапись, а не новую секцию:\n%s", lf.log.String())
|
||||
}
|
||||
}
|
||||
|
||||
// Список непокрытых секций переживает отказ разбора, то есть имя уже записано в
|
||||
// учёт. Промолчи здесь — и следующая доставка сочтёт его виденным, а событие не
|
||||
// вернётся ничем.
|
||||
func TestFoldОтказРазбораНазываетНовуюСекцию(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// Слой не выводится: плотных метрик в теле нет вовсе.
|
||||
body := withSection(t, fixture(t, "sparse_sleep.json"), "cycleTracking")
|
||||
deliver(t, lf.arch, lf.st, "d1", "Default", "auto-1", body)
|
||||
if _, err := lf.svc.Fold(ctx, "d1"); err == nil {
|
||||
t.Fatal("свёртка с неопределимым слоем прошла успешно")
|
||||
}
|
||||
|
||||
var found bool
|
||||
for _, rec := range logRecords(t, lf.log) {
|
||||
for _, name := range newSections(t, rec) {
|
||||
if name == "cycleTracking" {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Errorf("запись об отказе не назвала новую секцию:\n%s", lf.log.String())
|
||||
}
|
||||
|
||||
// И имя доехало до учёта — иначе терять было бы нечего.
|
||||
d, err := lf.st.LastDelivery(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("чтение доставки: %v", err)
|
||||
}
|
||||
if !strings.Contains(d.UncoveredSections, "cycleTracking") {
|
||||
t.Errorf("список в учёте %q не содержит секции", d.UncoveredSections)
|
||||
}
|
||||
}
|
||||
|
||||
// Пересборка проигрывает журнал заново, и повторная свёртка обязана дать тот же
|
||||
// состав событий: признак новизны — функция журнала, а не числа прогонов.
|
||||
func TestFoldПовторнаяСвёрткаДаётТотЖеСоставСобытий(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
body := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||||
|
||||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", body)
|
||||
first, err := lf.svc.Fold(ctx, "d1")
|
||||
if err != nil {
|
||||
t.Fatalf("первая свёртка: %v", err)
|
||||
}
|
||||
second, err := lf.svc.Fold(ctx, "d1")
|
||||
if err != nil {
|
||||
t.Fatalf("повторная свёртка: %v", err)
|
||||
}
|
||||
|
||||
if len(first.UncoveredNew) != len(second.UncoveredNew) {
|
||||
t.Fatalf("новых секций %v против %v", first.UncoveredNew, second.UncoveredNew)
|
||||
}
|
||||
for i := range first.UncoveredNew {
|
||||
if first.UncoveredNew[i] != second.UncoveredNew[i] {
|
||||
t.Errorf("на месте %d %q против %q", i, first.UncoveredNew[i], second.UncoveredNew[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Отказ СЛИЯНИЯ пишет имя в учёт ровно так же, как отказ разбора, — значит и
|
||||
// событие терять ему нельзя. Ветвь отдельная, и однажды она уже теряла его:
|
||||
// residue собирался вручную и новизну не нёс.
|
||||
func TestFoldОтказСлиянияНазываетНовуюСекцию(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
lf := newLoggedFold(t)
|
||||
ctx := context.Background()
|
||||
|
||||
breakMerge(t, lf.dbPath)
|
||||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1",
|
||||
withSection(t, fixture(t, "minute.json"), "ecg"))
|
||||
if _, err := lf.svc.Fold(ctx, "d1"); err == nil {
|
||||
t.Fatal("свёртка без таблицы объектов прошла успешно")
|
||||
}
|
||||
|
||||
var told bool
|
||||
for _, rec := range logRecords(t, lf.log) {
|
||||
for _, name := range newSections(t, rec) {
|
||||
if name == "ecg" {
|
||||
told = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if !told {
|
||||
t.Errorf("запись об отказе слияния не назвала новую секцию:\n%s", lf.log.String())
|
||||
}
|
||||
|
||||
// И событие действительно потеряно быть не могло: имя уже в учёте, то есть
|
||||
// следующая доставка сочтёт его виденным.
|
||||
d, err := lf.st.LastDelivery(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("чтение доставки: %v", err)
|
||||
}
|
||||
if !strings.Contains(d.UncoveredSections, "ecg") {
|
||||
t.Errorf("список в учёте %q не содержит секции — терять было бы нечего", d.UncoveredSections)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,243 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// UncoveredSection — строка перечня непокрытых секций: имя, которое поток
|
||||
// приносил, и границы его встреч по журналу.
|
||||
//
|
||||
// Число доставок и границы считаются по ВСЕМ статусам разбора: список непокрытых
|
||||
// секций переживает отказ свёртки, и молчать о таком имени значило бы терять как
|
||||
// раз подозрительное.
|
||||
type UncoveredSection struct {
|
||||
// Name — верхнеуровневый ключ `data`, дословно как прислал HAE. Приходит из
|
||||
// чужого тела: длина ограничена разбором, содержимое — ничем.
|
||||
Name string
|
||||
// Deliveries — сколько доставок принесло это имя.
|
||||
Deliveries int64
|
||||
// FirstSeen и FirstDeliveryID — первая встреча ПО ЖУРНАЛУ; LastSeen и
|
||||
// LastDeliveryID — последняя. Идентификатор нужен, чтобы достать тело из
|
||||
// архива и посмотреть форму секции глазами.
|
||||
FirstSeen time.Time
|
||||
FirstDeliveryID string
|
||||
LastSeen time.Time
|
||||
LastDeliveryID string
|
||||
}
|
||||
|
||||
// usableList — условие «списку непокрытых секций есть что сказать».
|
||||
//
|
||||
// Отдельной константой, потому что стоит во всех трёх запросах и означает одно:
|
||||
// по умолчанию колонка держит `[]`, а строки с ним до json_each доходить не
|
||||
// должны.
|
||||
//
|
||||
// `json_valid` здесь не перестраховка. `json_each` над неразбираемым значением
|
||||
// отвечает ошибкой и валит ВЕСЬ запрос, а не пропускает одну строку (проверено
|
||||
// на закреплённом драйвере: `SQL logic error: malformed JSON`). Одна такая
|
||||
// строка — из ручной правки, из будущей миграции данных мимо `json.Marshal`,
|
||||
// из восстановления базы чужим инструментом — обесценила бы и сверку (вечное
|
||||
// «сверка не состоялась» на каждой доставке), и перечень (полный отказ команды
|
||||
// без указания виновника). Штатный путь записи такого значения не производит, и
|
||||
// именно поэтому отказ был бы необъясним.
|
||||
const usableList = `uncovered_sections <> '[]' AND uncovered_sections <> ''`
|
||||
|
||||
// listOf — содержимое колонки, приведённое к разбираемому виду.
|
||||
//
|
||||
// Проверка стоит ВНУТРИ аргумента `json_each`, а не условием в `WHERE`:
|
||||
// табличная функция получает значение строки раньше, чем применится фильтр, и
|
||||
// порядок этот SQLite не обещает. Условие в `WHERE` работало бы, пока
|
||||
// планировщик проталкивает его вниз, и молча перестало бы — с ошибкой, роняющей
|
||||
// весь запрос, а не строку.
|
||||
const listOf = `json_each(CASE WHEN json_valid(d.uncovered_sections)
|
||||
THEN d.uncovered_sections ELSE '[]' END)`
|
||||
|
||||
// SectionsSeenBefore отвечает, какие из имён уже встречались в доставках,
|
||||
// стоящих в журнале СТРОГО РАНЬШЕ указанной.
|
||||
//
|
||||
// Строгое сравнение пары `(received_at, id)` решает три вещи разом: собственная
|
||||
// строка доставки в счёт не идёт при любом порядке записи исхода; повторная
|
||||
// свёртка той же доставки даёт тот же ответ, поэтому пересборка воспроизводит те
|
||||
// же события; и судит имя порядок ЖУРНАЛА, а не порядок прогона — строка учёта
|
||||
// становится видимой воркеру только после записи тела, так что при конкурентном
|
||||
// приёме доставка может стать видимой после более новой.
|
||||
//
|
||||
// Индекса по `uncovered_sections` в схеме нет, поэтому проход по журналу
|
||||
// полный, и форма запроса выбрана ЗАМЕРОМ (числа и метод — в `design.md`
|
||||
// изменения `aktivnaya-proverka-novyh-sekcij`, решение 3; повторяется прогоном
|
||||
// `tmp/seenmeasure`). Здесь только правило, которое из замера следует:
|
||||
//
|
||||
// - ранний выход `LIMIT 1` срабатывает, лишь когда первая встреча имени лежит
|
||||
// БЛИЗКО К НАЧАЛУ журнала, — строки просматриваются от старых к новым.
|
||||
// Секция, приезжающая давно, стоит десятки микросекунд; секция, появившаяся
|
||||
// только что, — почти полный проход на каждой доставке, пока её не покроет
|
||||
// отдельная задача;
|
||||
// - имён больше одного спрашиваются ОДНИМ запросом: тридцать два запроса
|
||||
// подряд стоили секунду с лишним на доставку, одна выборка — столько же,
|
||||
// сколько один полный проход, независимо от числа имён.
|
||||
//
|
||||
// Вызывается ВНЕ транзакции записи: проход по растущему журналу внутри неё
|
||||
// удерживал бы блокировку, а конкурирующий приём отвечает `500` по доставке, чьё
|
||||
// тело уже на диске.
|
||||
func (s *Store) SectionsSeenBefore(ctx context.Context, names []string, before DeliveryRef) (map[string]struct{}, error) {
|
||||
unique := make([]string, 0, len(names))
|
||||
dedup := make(map[string]struct{}, len(names))
|
||||
for _, name := range names {
|
||||
if _, ok := dedup[name]; ok {
|
||||
continue
|
||||
}
|
||||
dedup[name] = struct{}{}
|
||||
unique = append(unique, name)
|
||||
}
|
||||
if len(unique) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
at := FormatTime(before.ReceivedAt)
|
||||
if len(unique) == 1 {
|
||||
return s.sectionSeenOnce(ctx, unique[0], at, before.ID)
|
||||
}
|
||||
return s.sectionsSeenBatch(ctx, unique, at, before.ID)
|
||||
}
|
||||
|
||||
// sectionSeenOnce спрашивает про одно имя с ранним выходом.
|
||||
func (s *Store) sectionSeenOnce(ctx context.Context, name, at, id string) (map[string]struct{}, error) {
|
||||
const q = `
|
||||
SELECT 1
|
||||
FROM delivery d, ` + listOf + ` j
|
||||
WHERE ` + usableList + `
|
||||
AND (d.received_at, d.id) < (?, ?)
|
||||
AND j.value = ?
|
||||
LIMIT 1`
|
||||
|
||||
var one int
|
||||
err := s.db.QueryRowxContext(ctx, q, at, id, name).Scan(&one)
|
||||
switch {
|
||||
case err == nil:
|
||||
return map[string]struct{}{name: {}}, nil
|
||||
case errors.Is(err, sql.ErrNoRows):
|
||||
return nil, nil
|
||||
default:
|
||||
// Имя в текст ошибки не попадает целиком: оно приходит из чужого тела, а
|
||||
// ошибка уходит в лог уровня выше DEBUG. Та же граница, что у координат
|
||||
// наблюдения категориального значения.
|
||||
return nil, fmt.Errorf("section seen before (%s): %w", clipCoord(name), err)
|
||||
}
|
||||
}
|
||||
|
||||
// sectionsSeenBatch спрашивает про все имена одним проходом по журналу.
|
||||
func (s *Store) sectionsSeenBatch(ctx context.Context, names []string, at, id string) (map[string]struct{}, error) {
|
||||
q := `
|
||||
SELECT DISTINCT j.value
|
||||
FROM delivery d, ` + listOf + ` j
|
||||
WHERE ` + usableList + `
|
||||
AND (d.received_at, d.id) < (?, ?)
|
||||
AND j.value IN (?` + strings.Repeat(", ?", len(names)-1) + `)`
|
||||
|
||||
args := make([]any, 0, len(names)+2)
|
||||
args = append(args, at, id)
|
||||
for _, name := range names {
|
||||
args = append(args, name)
|
||||
}
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("sections seen before: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
seen := make(map[string]struct{}, len(names))
|
||||
for rows.Next() {
|
||||
var name string
|
||||
if err := rows.Scan(&name); err != nil {
|
||||
return nil, fmt.Errorf("scan section seen before: %w", err)
|
||||
}
|
||||
seen[name] = struct{}{}
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("sections seen before: %w", err)
|
||||
}
|
||||
return seen, nil
|
||||
}
|
||||
|
||||
// UncoveredSections возвращает перечень непокрытых секций — не больше limit
|
||||
// строк — и общее число различных имён в журнале.
|
||||
//
|
||||
// Предел объявлен, а не подразумевается: граница разбора в 32 имени действует на
|
||||
// ОДНУ доставку, а различных имён журнал накопит сколько угодно — достаточно
|
||||
// версии HAE, кладущей в ключ переменную часть. Второе возвращаемое значение
|
||||
// нужно, чтобы вывод мог назвать остаток числом, а не молча оборваться.
|
||||
//
|
||||
// Границы встреч берутся минимумом и максимумом склейки `received_at` и `id`:
|
||||
// метка хранится в RFC3339 фиксированной ширины, поэтому склейка сравнивается
|
||||
// лексикографически ровно как пара, а одна выборка с двумя агрегатами не
|
||||
// опирается на то, из какой строки SQLite возьмёт голые колонки.
|
||||
func (s *Store) UncoveredSections(ctx context.Context, limit int) ([]UncoveredSection, int64, error) {
|
||||
if limit <= 0 {
|
||||
return nil, 0, fmt.Errorf("предел строк перечня должен быть положительным, дано %d", limit)
|
||||
}
|
||||
|
||||
const countQ = `
|
||||
SELECT count(DISTINCT j.value)
|
||||
FROM delivery d, ` + listOf + ` j
|
||||
WHERE ` + usableList
|
||||
|
||||
var total int64
|
||||
if err := s.db.GetContext(ctx, &total, countQ); err != nil {
|
||||
return nil, 0, fmt.Errorf("count uncovered sections: %w", err)
|
||||
}
|
||||
|
||||
const q = `
|
||||
SELECT j.value,
|
||||
count(DISTINCT d.id),
|
||||
min(d.received_at || ' ' || d.id),
|
||||
max(d.received_at || ' ' || d.id)
|
||||
FROM delivery d, ` + listOf + ` j
|
||||
WHERE ` + usableList + `
|
||||
GROUP BY j.value
|
||||
ORDER BY j.value
|
||||
LIMIT ?`
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, q, limit)
|
||||
if err != nil {
|
||||
return nil, 0, fmt.Errorf("select uncovered sections: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var out []UncoveredSection
|
||||
for rows.Next() {
|
||||
var u UncoveredSection
|
||||
var first, last string
|
||||
if err := rows.Scan(&u.Name, &u.Deliveries, &first, &last); err != nil {
|
||||
return nil, 0, fmt.Errorf("scan uncovered section: %w", err)
|
||||
}
|
||||
if u.FirstSeen, u.FirstDeliveryID, err = splitJournalKey(first); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
if u.LastSeen, u.LastDeliveryID, err = splitJournalKey(last); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
out = append(out, u)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, 0, fmt.Errorf("select uncovered sections: %w", err)
|
||||
}
|
||||
return out, total, nil
|
||||
}
|
||||
|
||||
// splitJournalKey разбирает склейку метки журнала и идентификатора доставки.
|
||||
func splitJournalKey(key string) (time.Time, string, error) {
|
||||
at, id, ok := strings.Cut(key, " ")
|
||||
if !ok {
|
||||
return time.Time{}, "", fmt.Errorf("ключ журнала без разделителя: %q", key)
|
||||
}
|
||||
t, err := ParseTime(at)
|
||||
if err != nil {
|
||||
return time.Time{}, "", err
|
||||
}
|
||||
return t, id, nil
|
||||
}
|
||||
@@ -0,0 +1,268 @@
|
||||
package store_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.vakhrushev.me/av/healthlog/internal/store"
|
||||
|
||||
_ "modernc.org/sqlite" // прямая запись в файл базы — ради теста на битую колонку
|
||||
)
|
||||
|
||||
// seedUncovered заводит доставку с уже записанным исходом разбора: список
|
||||
// непокрытых секций живёт в учётной записи, и заполняет его свёртка.
|
||||
func seedUncovered(t *testing.T, st *store.Store, id string, at time.Time, status string, sections ...string) {
|
||||
t.Helper()
|
||||
|
||||
seedPending(t, st, id, at)
|
||||
err := st.FinishParse(context.Background(), id, store.ParseOutcome{
|
||||
Status: status,
|
||||
Layer: "hour",
|
||||
Uncovered: sections,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("исход разбора %q: %v", id, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Признак новизны обязан быть функцией ЖУРНАЛА, а не порядка свёртки: имя
|
||||
// считается виденным, только если встречалось строго раньше по паре
|
||||
// `(received_at, id)`.
|
||||
func TestSectionsSeenBeforeСудитПоЖурналу(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
|
||||
early := at
|
||||
late := at.Add(time.Second)
|
||||
seedUncovered(t, st, "d-early", early, store.ParsePartial, "ecg")
|
||||
seedUncovered(t, st, "d-late", late, store.ParsePartial, "ecg")
|
||||
|
||||
seen, err := st.SectionsSeenBefore(ctx, []string{"ecg"}, store.DeliveryRef{ID: "d-late", ReceivedAt: late})
|
||||
if err != nil {
|
||||
t.Fatalf("SectionsSeenBefore: %v", err)
|
||||
}
|
||||
if _, ok := seen["ecg"]; !ok {
|
||||
t.Error("для поздней доставки имя обязано быть виденным: раньше неё в журнале оно есть")
|
||||
}
|
||||
|
||||
// Ранняя доставка судится тем же запросом и обязана признать имя новым —
|
||||
// собственная строка в счёт не идёт, а более поздняя лежит после неё.
|
||||
seen, err = st.SectionsSeenBefore(ctx, []string{"ecg"}, store.DeliveryRef{ID: "d-early", ReceivedAt: early})
|
||||
if err != nil {
|
||||
t.Fatalf("SectionsSeenBefore: %v", err)
|
||||
}
|
||||
if _, ok := seen["ecg"]; ok {
|
||||
t.Error("для ранней доставки имя обязано быть новым: раньше неё его в журнале нет")
|
||||
}
|
||||
}
|
||||
|
||||
// Метка приёма хранится с секундной точностью, поэтому доставки одной секунды
|
||||
// разводятся идентификатором — иначе исход зависел бы от того, какая из них
|
||||
// свернулась первой.
|
||||
func TestSectionsSeenBeforeРазводитОднуСекундуИдентификатором(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
|
||||
seedUncovered(t, st, "d-a", at, store.ParsePartial, "symptoms")
|
||||
seedUncovered(t, st, "d-b", at, store.ParsePartial, "symptoms")
|
||||
|
||||
seen, err := st.SectionsSeenBefore(ctx, []string{"symptoms"}, store.DeliveryRef{ID: "d-b", ReceivedAt: at})
|
||||
if err != nil {
|
||||
t.Fatalf("SectionsSeenBefore: %v", err)
|
||||
}
|
||||
if _, ok := seen["symptoms"]; !ok {
|
||||
t.Error("для второй доставки той же секунды имя обязано быть виденным")
|
||||
}
|
||||
|
||||
seen, err = st.SectionsSeenBefore(ctx, []string{"symptoms"}, store.DeliveryRef{ID: "d-a", ReceivedAt: at})
|
||||
if err != nil {
|
||||
t.Fatalf("SectionsSeenBefore: %v", err)
|
||||
}
|
||||
if len(seen) != 0 {
|
||||
t.Errorf("для первой доставки той же секунды виденных имён быть не может, получено %v", seen)
|
||||
}
|
||||
}
|
||||
|
||||
// Повторный вопрос обязан давать тот же ответ: на этом стоит воспроизводимость
|
||||
// событий при пересборке журнала.
|
||||
func TestSectionsSeenBeforeИдемпотентен(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
|
||||
seedUncovered(t, st, "d1", at, store.ParsePartial, "ecg", "symptoms")
|
||||
seedUncovered(t, st, "d2", at.Add(time.Minute), store.ParsePartial, "ecg")
|
||||
|
||||
ref := store.DeliveryRef{ID: "d2", ReceivedAt: at.Add(time.Minute)}
|
||||
first, err := st.SectionsSeenBefore(ctx, []string{"ecg", "medications"}, ref)
|
||||
if err != nil {
|
||||
t.Fatalf("SectionsSeenBefore: %v", err)
|
||||
}
|
||||
second, err := st.SectionsSeenBefore(ctx, []string{"ecg", "medications"}, ref)
|
||||
if err != nil {
|
||||
t.Fatalf("SectionsSeenBefore: %v", err)
|
||||
}
|
||||
if len(first) != 1 || len(second) != 1 {
|
||||
t.Fatalf("виденных имён %d и %d, ожидалось по одному", len(first), len(second))
|
||||
}
|
||||
if _, ok := first["ecg"]; !ok {
|
||||
t.Error("`ecg` обязан быть виденным")
|
||||
}
|
||||
if _, ok := second["medications"]; ok {
|
||||
t.Error("`medications` журнал не приносил и виденным быть не может")
|
||||
}
|
||||
}
|
||||
|
||||
// Перечень отвечает по всем статусам разбора: список непокрытых секций
|
||||
// переживает отказ свёртки, и молчать о таком имени значило бы терять как раз
|
||||
// подозрительное.
|
||||
func TestUncoveredSectionsСчитаетВстречиИГраницы(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
|
||||
seedUncovered(t, st, "d1", at, store.ParsePartial, "symptoms")
|
||||
seedUncovered(t, st, "d2", at.Add(time.Minute), store.ParsePartial, "symptoms", "ecg")
|
||||
// Отказавшая доставка: имя записано, статус — `failed`.
|
||||
seedUncovered(t, st, "d3", at.Add(2*time.Minute), store.ParseFailed, "ecg")
|
||||
|
||||
got, total, err := st.UncoveredSections(ctx, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("UncoveredSections: %v", err)
|
||||
}
|
||||
if total != 2 {
|
||||
t.Errorf("различных имён %d, ожидалось 2", total)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("строк перечня %d, ожидалось 2", len(got))
|
||||
}
|
||||
|
||||
// Порядок детерминирован — по имени.
|
||||
if got[0].Name != "ecg" || got[1].Name != "symptoms" {
|
||||
t.Fatalf("порядок строк %q, %q", got[0].Name, got[1].Name)
|
||||
}
|
||||
|
||||
ecg := got[0]
|
||||
if ecg.Deliveries != 2 {
|
||||
t.Errorf("`ecg` принесли %d доставок, ожидалось 2", ecg.Deliveries)
|
||||
}
|
||||
if ecg.FirstDeliveryID != "d2" || ecg.LastDeliveryID != "d3" {
|
||||
t.Errorf("границы `ecg`: %q…%q, ожидалось d2…d3", ecg.FirstDeliveryID, ecg.LastDeliveryID)
|
||||
}
|
||||
if !ecg.FirstSeen.Equal(at.Add(time.Minute)) || !ecg.LastSeen.Equal(at.Add(2*time.Minute)) {
|
||||
t.Errorf("метки `ecg`: %s…%s", ecg.FirstSeen, ecg.LastSeen)
|
||||
}
|
||||
|
||||
sym := got[1]
|
||||
if sym.Deliveries != 2 || sym.FirstDeliveryID != "d1" || sym.LastDeliveryID != "d2" {
|
||||
t.Errorf("`symptoms`: %d доставок, %q…%q", sym.Deliveries, sym.FirstDeliveryID, sym.LastDeliveryID)
|
||||
}
|
||||
}
|
||||
|
||||
// Предел объявлен, а остаток обязан быть назван числом: «здесь всё» и «здесь
|
||||
// часть» — разные ответы, и молчаливый обрыв делает их неотличимыми.
|
||||
func TestUncoveredSectionsОстатокНазванЧислом(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
|
||||
seedUncovered(t, st, "d1", at, store.ParsePartial, "aaa", "bbb", "ccc")
|
||||
|
||||
got, total, err := st.UncoveredSections(ctx, 2)
|
||||
if err != nil {
|
||||
t.Fatalf("UncoveredSections: %v", err)
|
||||
}
|
||||
if len(got) != 2 {
|
||||
t.Fatalf("строк перечня %d, ожидалось 2", len(got))
|
||||
}
|
||||
if total != 3 {
|
||||
t.Errorf("всего имён %d, ожидалось 3", total)
|
||||
}
|
||||
}
|
||||
|
||||
// Пустой журнал и журнал без непокрытых секций отвечают пустым перечнем, а не
|
||||
// отказом: колонка по умолчанию держит `[]`, и до json_each такие строки не
|
||||
// доходят.
|
||||
func TestUncoveredSectionsНаЧистомЖурнале(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
st := open(t)
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
|
||||
seedUncovered(t, st, "d1", at, store.ParseDone)
|
||||
|
||||
got, total, err := st.UncoveredSections(ctx, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("UncoveredSections: %v", err)
|
||||
}
|
||||
if len(got) != 0 || total != 0 {
|
||||
t.Errorf("перечень %v, всего %d — ожидалась пустота", got, total)
|
||||
}
|
||||
}
|
||||
|
||||
// Одна учётная запись с неразбираемым содержимым колонки не имеет права
|
||||
// обесценить и сверку, и перечень: `json_each` над таким значением отвечает
|
||||
// ошибкой и валит ВЕСЬ запрос, а не пропускает строку.
|
||||
//
|
||||
// Штатный путь записи такого значения не производит — тем хуже был бы отказ:
|
||||
// он необъясним, а последствие вечно (сверка на каждой доставке говорила бы
|
||||
// «не состоялась», команда отказывала бы целиком).
|
||||
func TestUncoveredSectionsПереживаетБитуюСтроку(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
dbPath := filepath.Join(dir, "healthlog.db")
|
||||
st, err := store.Open(dbPath)
|
||||
if err != nil {
|
||||
t.Fatalf("открытие базы: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = st.Close() })
|
||||
|
||||
ctx := context.Background()
|
||||
at := ts(t, "2026-08-01T12:00:00Z")
|
||||
seedUncovered(t, st, "d1", at, store.ParsePartial, "ecg")
|
||||
seedUncovered(t, st, "d2", at.Add(time.Minute), store.ParsePartial, "symptoms")
|
||||
|
||||
db, err := sql.Open("sqlite", "file:"+dbPath+"?_pragma=busy_timeout(5000)")
|
||||
if err != nil {
|
||||
t.Fatalf("прямое подключение: %v", err)
|
||||
}
|
||||
defer func() { _ = db.Close() }()
|
||||
if _, err := db.ExecContext(ctx,
|
||||
`UPDATE delivery SET uncovered_sections = '{не JSON' WHERE id = ?`, "d1"); err != nil {
|
||||
t.Fatalf("порча колонки: %v", err)
|
||||
}
|
||||
|
||||
got, total, err := st.UncoveredSections(ctx, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("перечень отказал целиком из-за одной строки: %v", err)
|
||||
}
|
||||
if total != 1 || len(got) != 1 || got[0].Name != "symptoms" {
|
||||
t.Errorf("перечень %v (всего %d), ожидалась одна уцелевшая секция", got, total)
|
||||
}
|
||||
|
||||
seen, err := st.SectionsSeenBefore(ctx, []string{"symptoms"},
|
||||
store.DeliveryRef{ID: "d3", ReceivedAt: at.Add(2 * time.Minute)})
|
||||
if err != nil {
|
||||
t.Fatalf("сверка отказала из-за одной строки: %v", err)
|
||||
}
|
||||
if _, ok := seen["symptoms"]; !ok {
|
||||
t.Error("уцелевшая строка перестала участвовать в сверке")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user