package replay_test import ( "context" "log/slog" "os" "path/filepath" "sort" "strings" "testing" "time" "git.vakhrushev.me/av/healthlog/internal/archive" "git.vakhrushev.me/av/healthlog/internal/fold" "git.vakhrushev.me/av/healthlog/internal/ident" "git.vakhrushev.me/av/healthlog/internal/replay" "git.vakhrushev.me/av/healthlog/internal/store" ) // item — одна доставка тестового журнала. type item struct { id string at time.Time automationID string aggregation string fixture string } func fixture(t *testing.T, name string) []byte { t.Helper() body, err := os.ReadFile(filepath.Join("..", "hae", "testdata", name)) if err != nil { t.Fatalf("фикстура %s: %v", name, err) } return body } func openStore(t *testing.T, path string) *store.Store { t.Helper() st, err := store.Open(path) if err != nil { t.Fatalf("база %s: %v", path, err) } t.Cleanup(func() { _ = st.Close() }) return st } func openArchive(t *testing.T, root string) *archive.Archive { t.Helper() arch, err := archive.New(root) if err != nil { t.Fatalf("архив: %v", err) } return arch } // live воспроизводит живой приём: тело в архив, строка учёта, свёртка — в том // порядке и тем кодом, каким это делает `internal/ingest`. func live(t *testing.T, arch *archive.Archive, st *store.Store, items []item) { t.Helper() f := fold.New(arch, st, 0, slog.New(slog.DiscardHandler)) for _, it := range items { writeBody(t, arch, st, it, fixture(t, it.fixture)) _, _ = f.Fold(context.Background(), it.id) } } func writeBody(t *testing.T, arch *archive.Archive, st *store.Store, it item, body []byte) { t.Helper() rawPath, err := arch.Write(it.id, it.at, body) if err != nil { t.Fatalf("запись в архив: %v", err) } if st == nil { return } err = st.CreateDelivery(context.Background(), store.Delivery{ ID: it.id, ReceivedAt: it.at, AutomationID: it.automationID, Aggregation: it.aggregation, Bytes: int64(len(body)), SHA256: "-", RawPath: rawPath, Headers: `{"x-test":["1"]}`, ParseStatus: store.ParsePending, }) if err != nil { t.Fatalf("запись доставки: %v", err) } } // run проигрывает журнал в свежую базу и возвращает отчёт вместе с ней. func run(t *testing.T, ctx context.Context, arch *archive.Archive, src *store.Store, out string) (replay.Report, *store.Store) { t.Helper() dst := openStore(t, out) rep, err := replay.Run(ctx, replay.Options{ Archive: arch, Source: src, Target: dst, Fold: fold.New(arch, dst, 0, slog.New(slog.DiscardHandler)), }) if err != nil { t.Fatalf("проигрывание: %v", err) } return rep, dst } func fingerprint(t *testing.T, st *store.Store) string { t.Helper() fp, err := st.Fingerprint(context.Background()) if err != nil { t.Fatalf("отпечаток: %v", err) } return fp } // journal собирает журнал из фикстур с монотонными идентификаторами. func journal(t *testing.T, fixtures ...string) []item { t.Helper() base := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) out := make([]item, 0, len(fixtures)) for i, f := range fixtures { out = append(out, item{ id: ident.NewID(), at: base.Add(time.Duration(i+1) * time.Second), automationID: "auto-1", aggregation: "Default", fixture: f, }) } return out } // Главная проверка задачи: пересборка с нуля даёт то же состояние, что // накопленный приём, а повторный прогон ничего не меняет. func TestПересборкаСовпадаетСПриёмом(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json", "hour.json", "raw.json", "mixed.json", "sparse_sleep.json") live(t, arch, src, items) rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Bodies != len(items) { t.Fatalf("тел в журнале %d, ожидалось %d", rep.Bodies, len(items)) } if rep.Folded == 0 { t.Fatal("ни одна доставка не свернулась") } if rep.Adopted != 0 || rep.Orphans != 0 || rep.SkippedFiles != 0 { t.Errorf("журнал не должен был дать подобранных/сирот/пропусков: %+v", rep) } want := fingerprint(t, src) if rep.Fingerprint != want { t.Errorf("отпечаток пересобранной витрины не совпал с накопленной:\n приём %s\n пересборка %s", want, rep.Fingerprint) } // Повторный прогон в ещё одну базу обязан дать то же самое: победитель // координаты — функция множества кандидатов, а не порядка прихода. again, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild2.db")) if again.Fingerprint != rep.Fingerprint { t.Errorf("повторная пересборка изменила состояние:\n %s\n %s", rep.Fingerprint, again.Fingerprint) } } // Порядок проигрывания задаётся журналом, а не раскладкой файлов: доставка без // плотных метрик наследует слой ПРЕДШЕСТВУЮЩЕЙ доставки той же автоматизации. // // Журнал устроен так, что порядок имён файлов ОБРАТЕН хронологии. Проигрывание // по каталогу поставило бы доставку без плотных метрик первой — наследовать ей // было бы не от чего, слой не вывелся бы, и точки не сохранились бы вовсе. func TestПорядокЗадаётсяЖурналомАНеКаталогом(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() ids := []string{ident.NewID(), ident.NewID(), ident.NewID()} sort.Strings(ids) base := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) // Хронология обратна лексикографике имён: самый ранний файл — самая поздняя // доставка. items := []item{ {id: ids[2], at: base.Add(1 * time.Second), automationID: "a", aggregation: "Default", fixture: "hour.json"}, {id: ids[1], at: base.Add(2 * time.Second), automationID: "a", aggregation: "Default", fixture: "minute.json"}, {id: ids[0], at: base.Add(3 * time.Second), automationID: "a", aggregation: "Default", fixture: "sparse_sleep.json"}, } for _, it := range items { writeBody(t, arch, src, it, fixture(t, it.fixture)) } rep, dst := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.FailedLayer+rep.FailedMalformed+rep.FailedOther != 0 { t.Fatalf("отказов %d: доставка без плотных метрик не нашла предшественника — порядок взят из каталога", rep.FailedLayer+rep.FailedMalformed+rep.FailedOther) } // Предшественник — минутная доставка, значит эпизоды сна легли в minute. hours, err := dst.BucketHours(ctx, "sleep_analysis", "minute") if err != nil { t.Fatalf("часы объектов: %v", err) } if len(hours) == 0 { t.Error("эпизоды сна не унаследовали слой предшествующей доставки") } } // Тело без учётной записи — не экзотика: приём кладёт тело на диск раньше // строки в базе, и отказ на вставке оставляет тело без учёта. func TestТелоБезУчётаПодбирается(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() orphanBody := fixture(t, "minute.json") orphan := item{id: ident.NewID(), at: time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)} // Строки учёта нет — только тело. writeBody(t, arch, nil, orphan, orphanBody) rep, dst := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Adopted != 1 { t.Fatalf("подобрано %d тел, ожидалось 1: %+v", rep.Adopted, rep) } if rep.Folded != 1 { t.Fatalf("свёрнуто %d, ожидалась 1", rep.Folded) } if rep.Buckets == 0 { t.Error("точки подобранного тела не доехали до витрины") } // Размер и хеш считаются по РАСПАКОВАННОМУ телу — как их считает приём. got, err := dst.DeliveryForParse(ctx, orphan.id) if err != nil { t.Fatalf("учёт подобранного тела: %v", err) } wantAt, err := ident.TimeOf(orphan.id) if err != nil { t.Fatalf("время из ULID: %v", err) } if !got.ReceivedAt.Equal(wantAt) { t.Errorf("метка приёма %v, ожидалась из ULID %v", got.ReceivedAt, wantAt) } all, err := dst.ListDeliveries(ctx) if err != nil { t.Fatalf("учёт: %v", err) } if len(all) != 1 || all[0].Bytes != int64(len(orphanBody)) { t.Errorf("размер подобранного тела %v, ожидался по распакованному %d", all, len(orphanBody)) } } // Учётная запись без тела станет штатной, когда появится ретеншен архива: // тела срезаются до даты проверенного экспорта, а строки живут дольше. func TestУчётБезТелаНеРоняетПрогон(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json") live(t, arch, src, items) // Строка есть, тело исчезло. if err := os.Remove(filepath.Join(arch.Root(), "2026", "08", "01", items[0].id+".json.gz")); err != nil { t.Fatalf("удаление тела: %v", err) } rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Orphans != 1 { t.Errorf("записей без тела %d, ожидалась 1: %+v", rep.Orphans, rep) } if rep.Bodies != 0 { t.Errorf("тел %d, ожидался 0", rep.Bodies) } } // Файл, телом не являющийся, считается отдельно: молчаливый пропуск означал бы // «тело есть, а в отчёте его нет». func TestФайлНеТелоСчитаетсяОтдельно(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json") live(t, arch, src, items) day := filepath.Join(arch.Root(), "2026", "08", "01") // Остаток прерванной записи и файл с именем, которое не идентификатор. for _, name := range []string{items[0].id + ".json.gz.tmp", "readme.txt", "не-ulid.json.gz"} { if err := os.WriteFile(filepath.Join(day, name), []byte("x"), 0o600); err != nil { t.Fatalf("подготовка файла: %v", err) } } rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.SkippedFiles != 3 { t.Errorf("пропущено файлов %d, ожидалось 3: %+v", rep.SkippedFiles, rep) } if rep.Bodies != 1 || rep.Folded != 1 { t.Errorf("тел %d, свёрнуто %d, ожидалось 1 и 1", rep.Bodies, rep.Folded) } } // Слой прошлого разбора не должен доживать до наследования: иначе витрина // оказывается функцией предыдущего прогона, а не журнала. func TestСлойПрошлогоРазбораНеНаследуется(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json", "hour.json") live(t, arch, src, items) clean, _ := run(t, ctx, arch, src, filepath.Join(dir, "clean.db")) // Портим производное поле: как если бы прежний разбор вывел другой слой. for _, it := range items { err := src.FinishParse(ctx, it.id, store.ParseOutcome{ Status: store.ParseDone, Layer: "raw", }) if err != nil { t.Fatalf("порча derived_layer: %v", err) } } dirty, _ := run(t, ctx, arch, src, filepath.Join(dir, "dirty.db")) if dirty.Fingerprint != clean.Fingerprint { t.Errorf("слой прошлого разбора повлиял на пересборку:\n чистая %s\n с порчей %s", clean.Fingerprint, dirty.Fingerprint) } } // Отмена прекращает проигрывание: это требование прекратить работу, а не // свойство доставки. func TestОтменаПрекращаетПроигрывание(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) items := journal(t, "minute.json", "hour.json", "raw.json") live(t, arch, src, items) ctx, cancel := context.WithCancel(context.Background()) defer cancel() dst := openStore(t, filepath.Join(dir, "rebuild.db")) rep, err := replay.Run(ctx, replay.Options{ Archive: arch, Source: src, Target: dst, Fold: fold.New(arch, dst, 0, slog.New(slog.DiscardHandler)), // Отмена приходит посреди журнала — так же, как её принесёт сигнал. Progress: func(done, _ int) { if done == 1 { cancel() } }, }) if err != nil { t.Fatalf("проигрывание: %v", err) } if !rep.Canceled { t.Error("отмена не отмечена в отчёте") } if rep.Folded != 1 { t.Errorf("свёрнуто %d доставок, ожидалась 1 до отмены", rep.Folded) } // Отмена до начала работы — тоже отмена, а не отказ. stopped, stop := context.WithCancel(context.Background()) stop() early, err := replay.Run(stopped, replay.Options{ Archive: arch, Source: src, Target: openStore(t, filepath.Join(dir, "rebuild2.db")), Fold: fold.New(arch, dst, 0, slog.New(slog.DiscardHandler)), }) if err != nil { t.Fatalf("проигрывание при отменённом контексте: %v", err) } if !early.Canceled || early.Folded != 0 { t.Errorf("отмена до старта дала %+v", early) } } // Нечитаемый или отсутствующий каталог архива — отказ, а не пустой журнал: // пустая витрина совпадает по отпечатку с пустой витриной и выглядит идеальной // сходимостью. func TestОтсутствующийАрхивЭтоОтказ(t *testing.T) { t.Parallel() dir := t.TempDir() if _, err := archive.Existing(filepath.Join(dir, "нет-такого")); err == nil { t.Fatal("отсутствующий каталог архива не дал отказа") } arch := openArchive(t, filepath.Join(dir, "raw")) if err := os.RemoveAll(arch.Root()); err != nil { t.Fatalf("удаление каталога: %v", err) } dst := openStore(t, filepath.Join(dir, "rebuild.db")) _, err := replay.Run(context.Background(), replay.Options{ Archive: arch, Target: dst, Fold: fold.New(arch, dst, 0, slog.New(slog.DiscardHandler)), }) if err == nil { t.Error("исчезнувший каталог архива дал пустой журнал вместо отказа") } } // Битое тело не срывает прогон: остальные доставки обязаны проиграться. func TestБитоеТелоНеСрываетПрогон(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json", "hour.json") live(t, arch, src, items) broken := filepath.Join(arch.Root(), "2026", "08", "01", items[0].id+".json.gz") if err := os.WriteFile(broken, []byte("не gzip"), 0o600); err != nil { t.Fatalf("порча тела: %v", err) } rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) // Битый gzip — «прочее», а не невыведенный слой: классы разведены именно // затем, чтобы человек не искал дефект там, где его нет. if rep.FailedOther != 1 { t.Errorf("прочих отказов %d, ожидался 1: %+v", rep.FailedOther, rep) } if rep.Folded != 1 { t.Errorf("свёрнуто %d, ожидалась 1 — прогон сорвался на битом теле", rep.Folded) } } // Одно имя тела в двух каталогах — копия, восстановленная руками, или тело, // переложенное не туда. Вторая запись журнала с тем же идентификатором сорвала // бы весь прогон отказом по первичному ключу. func TestПовторИдентификатораНеСрываетПрогон(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json") live(t, arch, src, items) // Тот же файл, но под другой датой. body, err := os.ReadFile(filepath.Join(arch.Root(), "2026", "08", "01", items[0].id+".json.gz")) if err != nil { t.Fatalf("чтение тела: %v", err) } other := filepath.Join(arch.Root(), "2026", "07", "31") if err := os.MkdirAll(other, 0o755); err != nil { t.Fatalf("каталог: %v", err) } if err := os.WriteFile(filepath.Join(other, items[0].id+".json.gz"), body, 0o600); err != nil { t.Fatalf("копия тела: %v", err) } rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Duplicates != 1 { t.Errorf("повторов %d, ожидался 1: %+v", rep.Duplicates, rep) } if rep.Folded != 1 { t.Errorf("свёрнуто %d, ожидалась 1 — повтор сорвал прогон", rep.Folded) } } // Доставки одной секунды упорядочиваются идентификатором: `received_at` хранится // с секундной точностью, и без второго ключа два прогона одного журнала могли бы // разойтись. func TestДоставкиОднойСекундыУпорядоченыИдентификатором(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() ids := []string{ident.NewID(), ident.NewID()} sort.Strings(ids) at := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) // Обе доставки — одна секунда. Первая по идентификатору минутная, вторая без // плотных метрик: если тай-брейк исчезнет, вторая может пойти первой и // остаться без предшественника. writeBody(t, arch, src, item{id: ids[0], at: at, automationID: "a"}, fixture(t, "minute.json")) writeBody(t, arch, src, item{id: ids[1], at: at, automationID: "a"}, fixture(t, "sparse_sleep.json")) rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.FailedLayer != 0 { t.Fatalf("слой не вывелся у %d доставок: порядок внутри секунды не задан", rep.FailedLayer) } // Повтор в другую базу обязан дать тот же отпечаток. again, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild2.db")) if again.Fingerprint != rep.Fingerprint { t.Errorf("порядок внутри секунды не детерминирован:\n %s\n %s", rep.Fingerprint, again.Fingerprint) } } // Учётная запись, тела которой нет, переносится в базу назначения: не перенести // значило бы стереть первой же подменой единственное свидетельство, что // доставка была, — тела уже нет, восстановить нечем. func TestУчётБезТелаПереноситсяВБазуНазначения(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json", "hour.json") live(t, arch, src, items) if err := os.Remove(filepath.Join(arch.Root(), "2026", "08", "01", items[0].id+".json.gz")); err != nil { t.Fatalf("удаление тела: %v", err) } rep, dst := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Orphans != 1 { t.Fatalf("записей без тела %d, ожидалась 1", rep.Orphans) } got, err := dst.ListDeliveries(ctx) if err != nil { t.Fatalf("учёт: %v", err) } if len(got) != len(items) { t.Fatalf("строк учёта %d, ожидалось %d — запись без тела потеряна", len(got), len(items)) } // Заголовки переносятся дословно: в архиве их нет вовсе. for _, d := range got { if d.Headers != `{"x-test":["1"]}` { t.Errorf("заголовки доставки %s не дошли дословно: %q", d.ID, d.Headers) } } // Статус — `failed`, а не `pending`. Различие несущее: `pending` означает // «этим разбором ещё не смотрели» и обещает данные, которых не появится — // тела уже нет, — а ретеншен, который pending не трогает никогда, берёг бы // такие строки вечно. b, err := dst.DeliveryStatus(ctx, items[0].id) if err != nil { t.Fatalf("статус записи без тела: %v", err) } if b != store.ParseFailed { t.Errorf("статус записи без тела %q, ожидался %q", b, store.ParseFailed) } } // Имя тела обязано быть КАНОНИЧЕСКИМ идентификатором. Разбор с приведением // (обрезка пробелов, регистр) дал бы одну координату журнала двум файлам, и // подложенный вытеснил бы настоящий — невидимые пробелы в этих данных уже // встречались. func TestИмяТелаОбязаноБытьКаноническим(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json") live(t, arch, src, items) day := filepath.Join(arch.Root(), "2026", "08", "01") body, err := os.ReadFile(filepath.Join(day, items[0].id+".json.gz")) if err != nil { t.Fatalf("чтение тела: %v", err) } // Те же 26 знаков, но с невидимым префиксом и в верхнем регистре: обе формы // ident.Parse приводит к тому же идентификатору. for _, name := range []string{" " + items[0].id + ".json.gz", strings.ToUpper(items[0].id) + ".json.gz"} { if err := os.WriteFile(filepath.Join(day, name), body, 0o600); err != nil { t.Fatalf("подложенный файл: %v", err) } } rep, _ := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Bodies != 1 { t.Errorf("тел %d, ожидалось 1: подложенное имя принято за тело (%+v)", rep.Bodies, rep) } if rep.SkippedFiles != 2 { t.Errorf("пропущено %d, ожидалось 2", rep.SkippedFiles) } if rep.Duplicates != 0 { t.Errorf("повторов %d: настоящее тело вытеснено подложенным", rep.Duplicates) } } // Число пропущенных сущностей — производное от разбора поле, и пересборка его // не переносит. Пустота у него значит «не измерялось», а перенесённое число // выдавало бы измерение ПРЕЖНЕГО разбора за измерение текущего — притом что по // нему принимается необратимое решение об удалении тела. func TestЧислоПропусковНеПереноситсяВПересобраннуюБазу(t *testing.T) { t.Parallel() dir := t.TempDir() arch := openArchive(t, filepath.Join(dir, "raw")) src := openStore(t, filepath.Join(dir, "live.db")) ctx := context.Background() items := journal(t, "minute.json") live(t, arch, src, items) // Проставляем счётчик в исходной базе, как если бы его измерил прежний // разбор, и убираем тело: пересборке будет нечего пересчитывать. n := int64(7) for _, it := range items { err := src.FinishParse(ctx, it.id, store.ParseOutcome{ Status: store.ParseDone, SkippedEntities: &n, }) if err != nil { t.Fatalf("простановка счётчика: %v", err) } } // Тело убираем: доставка становится записью без тела, пересборка её не // сворачивает — и производные поля обязаны начаться пустыми, а не приехать // из журнала. raw := filepath.Join(dir, "raw") if err := os.RemoveAll(raw); err != nil { t.Fatalf("удаление тел: %v", err) } if err := os.MkdirAll(raw, 0o755); err != nil { t.Fatalf("пересоздание каталога архива: %v", err) } rep, dst := run(t, ctx, arch, src, filepath.Join(dir, "rebuild.db")) if rep.Orphans == 0 { t.Fatal("доставка не стала записью без тела — тест проверяет не то") } got, err := dst.LastDelivery(ctx) if err != nil { t.Fatalf("чтение доставки: %v", err) } if got.SkippedEntities != nil { t.Errorf("пересборка перенесла счётчик прежнего разбора: %d", *got.SkippedEntities) } }