Приём отвечает 200 до свёртки, свёртку ведёт фоновый воркер

- Очередью служит сама таблица: доставка ждёт свёртки в статусе `pending`,
  канал несёт только бит «есть работа». Переполнять нечего, падение процесса
  очередь не теряет, а подбор `pending` при старте — обычный проход воркера, а
  не отдельный код. Классификация исхода общая с пересборкой журнала.
- Исход разбора начал отражать доставку, а не обстоятельства: отмена и
  занятость базы статус не меняют (иначе конкуренция за базу выводила бы
  доставку из очереди навсегда), паника свёртки больше не валит процесс, а
  учёт доставки идёт через транзакцию с повторами.
- Длинный бюджет ответа выдан маршруту приёма, а не всему серверу:
  `write_timeout` в Go покрывает и чтение тела, и общий подъём снял бы защиту с
  остальных маршрутов.

internal/fold/busy_test.go

@@ -0,0 +1,97 @@
+package fold_test
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"errors"
+	"flag"
+	"log/slog"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	_ "modernc.org/sqlite" // тот же чистый Go-драйвер, что и у хранилища
+
+	"git.vakhrushev.me/av/healthlog/internal/archive"
+	"git.vakhrushev.me/av/healthlog/internal/fold"
+	"git.vakhrushev.me/av/healthlog/internal/store"
+)
+
+// Прогон под удерживаемой блокировкой намеренно не входит в `task test` и
+// `task gate`: `busy_timeout` — пять секунд, повторов транзакции пять, то есть
+// один этот тест стоит около двадцати пяти секунд, а гейт гоняет тесты трижды
+// (обычно, на флаки и под детектором гонок).
+//
+// Проверяет он при этом центральное решение задачи «разнести ответ и свёртку»:
+// занятость базы — обстоятельство, а не свойство доставки, и доставка обязана
+// остаться в очереди. Ошибка здесь означает молчаливую потерю: `failed` фоновая
+// свёртка не подбирает никогда, а вернуть доставку может только пересборка с
+// остановкой сервиса и ручной подменой базы.
+var runBusy = flag.Bool("healthlog.busy", false,
+	"прогнать свёртку под удерживаемой блокировкой базы (около 25 секунд)")
+
+func TestBusyЗанятаяБазаОставляетДоставкуВОчереди(t *testing.T) {
+	if !*runBusy {
+		t.Skip("прогон под блокировкой выключен: задайте -healthlog.busy")
+	}
+
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "healthlog.db")
+
+	arch, err := archive.New(filepath.Join(dir, "raw"))
+	if err != nil {
+		t.Fatalf("архив: %v", err)
+	}
+	st, err := store.Open(dbPath)
+	if err != nil {
+		t.Fatalf("база: %v", err)
+	}
+	defer func() { _ = st.Close() }()
+
+	var logs bytes.Buffer
+	f := fold.New(arch, st, 0, slog.New(slog.NewJSONHandler(&logs, nil)))
+	deliver(t, arch, st, "d1", "Minutes", "auto-1", fixture(t, "minute.json"))
+
+	// Второе соединение держит запись, как её держит свёртка широкой доставки:
+	// измерено 11 секунд на 16 тысячах объектов, то есть окно реальное.
+	holder, err := sql.Open("sqlite", "file:"+dbPath+"?_pragma=busy_timeout(100)&_txlock=immediate")
+	if err != nil {
+		t.Fatalf("второе соединение: %v", err)
+	}
+	defer func() { _ = holder.Close() }()
+
+	tx, err := holder.BeginTx(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("удержание записи: %v", err)
+	}
+	if _, err := tx.Exec(`UPDATE delivery SET points = points WHERE id = 'd1'`); err != nil {
+		t.Fatalf("удержание записи: %v", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	_, err = f.Fold(context.Background(), "d1")
+	if !errors.Is(err, store.ErrBusy) {
+		t.Fatalf("ошибка свёртки = %v, ожидалась %v", err, store.ErrBusy)
+	}
+
+	status, err := st.DeliveryStatus(context.Background(), "d1")
+	if err != nil {
+		t.Fatalf("DeliveryStatus: %v", err)
+	}
+	if status != store.ParsePending {
+		t.Errorf("parse_status = %q, ожидался %q: занятость базы вывела доставку из очереди",
+			status, store.ParsePending)
+	}
+
+	// Статуса мало: пока база занята, запись `failed` тоже не проходит, и
+	// `pending` получился бы и без правила. Различает их лог — свёртка обязана
+	// сказать «отложено», а не «отказ».
+	out := logs.String()
+	if !strings.Contains(out, "delivery fold deferred") {
+		t.Errorf("нет записи об отложенной свёртке:\n%s", out)
+	}
+	if strings.Contains(out, "delivery fold failed") {
+		t.Errorf("занятость базы записана отказом доставки:\n%s", out)
+	}
+}

internal/fold/fold.go

@@ -90,13 +90,33 @@ type Stats struct {
 	UncoveredDropped int
 }
 
+// ErrPanicked — свёртка паниковала. Доставка получает `failed`: тело в архиве, и
+// пересборка вернёт её, когда дефект будет исправлен.
+var ErrPanicked = errors.New("свёртка паниковала")
+
 // Fold разбирает тело доставки и раскладывает точки по часовым объектам.
 //
 // Это единственный логирующий чекпоинт свёртки: транспорт и приём исход
 // разбора не логируют. Значения точек и имена устройств в лог не попадают —
 // данные о здоровье чувствительнее токенов.
-func (s *Service) Fold(ctx context.Context, deliveryID string) (Stats, error) {
-	var stats Stats
+//
+// Паника перехватывается ЗДЕСЬ, у той же границы, что пишет исход разбора.
+// Пока свёртка шла внутри HTTP-обработчика, панику ловил middleware.Recoverer и
+// она стоила одного ответа; из фоновой горутины она валит процесс целиком, а
+// `restart: unless-stopped` поднимает его снова — и первый же проход берёт ту
+// же доставку, то есть дефект превращается в цикл перезапуска, при котором
+// приём не работает вовсе. Перехват у этой границы, а не у вызывающего,
+// оставляет писателя `parse_status` единственным.
+func (s *Service) Fold(ctx context.Context, deliveryID string) (stats Stats, err error) {
+	defer func() {
+		r := recover()
+		if r == nil {
+			return
+		}
+		stats = Stats{}
+		err = fmt.Errorf("%w: %v", ErrPanicked, r) //nolint:errorlint // причину раскрываем текстом, sentinel — для ветвления
+		s.fail(ctx, deliveryID, err, nil)
+	}()
 
 	d, err := s.store.DeliveryForParse(ctx, deliveryID)
 	if err != nil {
@@ -282,9 +302,28 @@ func (s *Service) finish(ctx context.Context, deliveryID string, out store.Parse
 	return nil
 }
 
-// fail отмечает доставку неразобранной. Тело остаётся в архиве, и её подберёт
-// пересборка — приём при этом не затрагивается: сохранили значит приняли.
+// fail записывает исход неудачной свёртки.
+//
+// Исход отражает ДОСТАВКУ, а не обстоятельства. Отмена снаружи и занятость базы
+// работой доставки не являются: они означают «не сделано», а не «не выходит».
+// Статус в этих случаях не трогается вовсе — доставка остаётся `pending` и
+// подбирается следующим проходом. Иначе конкуренция за базу (после разнесения
+// ответа и свёртки она штатная) выводила бы доставку из очереди навсегда:
+// `failed` возвращает только пересборка, то есть ручная операция с остановкой
+// сервиса.
+//
+// Всё прочее — непонятое содержимое, невыводимый слой, нечитаемое или слишком
+// большое тело, исчерпанный дедлайн — свойства самой доставки, и повторять их
+// бесполезно: статус `failed`, тело ждёт пересборки. Приём при этом не
+// затрагивается: сохранили значит приняли.
 func (s *Service) fail(ctx context.Context, deliveryID string, cause error, uncovered []string) {
+	if store.Transient(cause) {
+		// WARN, а не ERROR: пройдёт само, разбирать нечего. Строка нужна, чтобы
+		// повтор не выглядел беспричинным.
+		s.log.WarnContext(ctx, "delivery fold deferred", "error", cause, "delivery_id", deliveryID)
+		return
+	}
+
 	level := slog.LevelError
 	switch {
 	case errors.Is(cause, hae.ErrLayerUnknown):

internal/fold/fold_test.go

@@ -324,3 +324,65 @@ func TestFoldПересвёрткаОчищаетСписок(t *testing.T) {
 		t.Errorf("статус %q, ожидался %q", d.ParseStatus, store.ParseDone)
 	}
 }
+
+// Отмена снаружи не превращается в свойство доставки: работа не сделана, но
+// доставка остаётся в очереди и будет свёрнута снова. Иначе остановка сервиса в
+// неудачный момент выводила бы доставку из очереди навсегда — `failed` фоновая
+// свёртка не подбирает никогда, и вернуть её могла бы только пересборка с
+// остановкой сервиса и ручной подменой базы.
+func TestFoldОтменаОставляетДоставкуВОчереди(t *testing.T) {
+	t.Parallel()
+
+	f, arch, st := newFold(t)
+	deliver(t, arch, st, "d1", "Minutes", "auto-1", fixture(t, "minute.json"))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	if _, err := f.Fold(ctx, "d1"); err == nil {
+		t.Fatal("свёртка на отменённом контексте прошла успешно")
+	}
+
+	status, err := st.DeliveryStatus(context.Background(), "d1")
+	if err != nil {
+		t.Fatalf("DeliveryStatus: %v", err)
+	}
+	if status != store.ParsePending {
+		t.Errorf("parse_status = %q, ожидался %q", status, store.ParsePending)
+	}
+
+	n, err := st.CountBuckets(context.Background())
+	if err != nil {
+		t.Fatalf("CountBuckets: %v", err)
+	}
+	if n != 0 {
+		t.Errorf("объектов %d: прерванная свёртка оставила половину", n)
+	}
+}
+
+// Непонятое содержимое, наоборот, свойство самой доставки: повторять её
+// бесполезно, и она выводится из очереди.
+func TestFoldНепонятоеСодержимоеВыводитИзОчереди(t *testing.T) {
+	t.Parallel()
+
+	f, arch, st := newFold(t)
+	ctx := context.Background()
+
+	// Метрика есть, но слой определить нечем: плотных метрик нет, заголовок
+	// ничего не означает, наследовать не от чего.
+	body := []byte(`{"data":{"metrics":[{"name":"m","units":"u","data":[` +
+		`{"date":"2026-07-31 12:00:00 +0300","qty":1}]}]}}`)
+	deliver(t, arch, st, "d1", "Default", "auto-1", body)
+
+	if _, err := f.Fold(ctx, "d1"); err == nil {
+		t.Fatal("свёртка непонятого содержимого прошла успешно")
+	}
+
+	status, err := st.DeliveryStatus(ctx, "d1")
+	if err != nil {
+		t.Fatalf("DeliveryStatus: %v", err)
+	}
+	if status != store.ParseFailed {
+		t.Errorf("parse_status = %q, ожидался %q", status, store.ParseFailed)
+	}
+}

internal/httpapi/httpapi.go

@@ -23,22 +23,28 @@ type Options struct {
 	Log         *slog.Logger
 	WriteTokens []string
 	MaxBodyMB   int
+	// IngestWriteBudget — сколько отводится маршруту приёма на чтение тела
+	// вместе с отправкой ответа. Ноль означает «полагаться на WriteTimeout
+	// сервера», и полагаться на него нельзя, см. handleIngest.
+	IngestWriteBudget time.Duration
 }
 
 type api struct {
-	ingest      *ingest.Service
-	log         *slog.Logger
-	writeTokens []string
-	maxBody     int64
+	ingest       *ingest.Service
+	log          *slog.Logger
+	writeTokens  []string
+	maxBody      int64
+	ingestBudget time.Duration
 }
 
 // New собирает HTTP-роутер.
 func New(o Options) http.Handler {
 	a := &api{
-		ingest:      o.Ingest,
-		log:         o.Log,
-		writeTokens: o.WriteTokens,
-		maxBody:     int64(o.MaxBodyMB) << 20,
+		ingest:       o.Ingest,
+		log:          o.Log,
+		writeTokens:  o.WriteTokens,
+		maxBody:      int64(o.MaxBodyMB) << 20,
+		ingestBudget: o.IngestWriteBudget,
 	}
 
 	r := chi.NewRouter()

internal/httpapi/httpapi_test.go

@@ -3,6 +3,7 @@ package httpapi_test
 import (
 	"bytes"
 	"compress/gzip"
+	"context"
 	"encoding/json"
 	"log/slog"
 	"net/http"
@@ -10,9 +11,9 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"git.vakhrushev.me/av/healthlog/internal/archive"
-	"git.vakhrushev.me/av/healthlog/internal/fold"
 	"git.vakhrushev.me/av/healthlog/internal/httpapi"
 	"git.vakhrushev.me/av/healthlog/internal/ingest"
 	"git.vakhrushev.me/av/healthlog/internal/store"
@@ -259,6 +260,28 @@ func TestHealthz(t *testing.T) {
 	}
 }
 
+// Транспорт, не умеющий дедлайнов, приём не роняет: цена отказа здесь наивысшая
+// в проекте — доставка, не попавшая в архив, не попадает и в журнал.
+func TestПриёмРаботаетНаТранспортеБезДедлайнов(t *testing.T) {
+	h, st := newAPI(t, nil)
+
+	req := httptest.NewRequest(http.MethodPost, "/api/v1/ingest",
+		strings.NewReader(`{"data":{"metrics":[]}}`))
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("статус = %d, ожидался 200", rec.Code)
+	}
+	n, err := st.CountDeliveries(context.Background())
+	if err != nil {
+		t.Fatalf("CountDeliveries: %v", err)
+	}
+	if n != 1 {
+		t.Errorf("доставок в учёте %d, ожидалась 1", n)
+	}
+}
+
 func newAPI(t *testing.T, writeTokens []string) (http.Handler, *store.Store) {
 	t.Helper()
 	dir := t.TempDir()
@@ -276,10 +299,13 @@ func newAPI(t *testing.T, writeTokens []string) (http.Handler, *store.Store) {
 
 	log := slog.New(slog.DiscardHandler)
 	h := httpapi.New(httpapi.Options{
-		Ingest:      ingest.New(arch, st, fold.New(arch, st, 0, log), log),
+		Ingest:      ingest.New(arch, st, nil, log),
 		Log:         log,
 		WriteTokens: writeTokens,
 		MaxBodyMB:   1,
+		// Бюджет задаётся всегда: httptest.ResponseRecorder дедлайнов не умеет,
+		// и это ровно тот транспорт, на котором приём обязан продолжать работать.
+		IngestWriteBudget: time.Minute,
 	})
 	return h, st
 }

internal/httpapi/ingest.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"time"
 
 	"git.vakhrushev.me/av/healthlog/internal/ingest"
 )
@@ -22,6 +23,8 @@ type ingestResponse struct {
 // Код ответа отражает ДОСТАВКУ, а не разбор: 200 означает «тело сохранено в
 // архив», и этого достаточно, потому что разобрать сохранённое можно всегда.
 func (a *api) handleIngest(w http.ResponseWriter, r *http.Request) {
+	a.extendWriteDeadline(w, r)
+
 	body, err := readBody(w, r, a.maxBody)
 	if err != nil {
 		writeReadError(w, err)
@@ -46,6 +49,31 @@ func (a *api) handleIngest(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
+// extendWriteDeadline даёт маршруту приёма собственный бюджет ответа.
+//
+// `WriteTimeout` сервера ставится в `readRequest`, то есть ДО вызова
+// обработчика, и потому покрывает не только запись ответа, но и чтение тела:
+// при `read_timeout` в пять минут и `write_timeout` в тридцать секунд загрузка
+// длиннее тридцати секунд обрывается, а `read_timeout` при этом обещает пять
+// минут. Обрывается молча — обработчик ошибки записи не видит, а `accessLog`
+// пишет `status_code=200`.
+//
+// Лечится это здесь, а не подъёмом общего таймаута: длинный бюджет нужен
+// одному маршруту, и подъём снял бы защиту от застрявшей записи со всех
+// остальных.
+//
+// Транспорт, не умеющий дедлайнов, отказом приёма не является: приём —
+// единственное место, где поток вообще существует, и терять доставку из-за
+// неподдержанной оптимизации нельзя.
+func (a *api) extendWriteDeadline(w http.ResponseWriter, r *http.Request) {
+	if a.ingestBudget <= 0 {
+		return
+	}
+	if err := http.NewResponseController(w).SetWriteDeadline(time.Now().Add(a.ingestBudget)); err != nil {
+		a.log.DebugContext(r.Context(), "write deadline not set", "error", err)
+	}
+}
+
 // metaFromHeaders достаёт то, что автоматизация Health Auto Export
 // рассказывает о себе своими заголовками. Именованные поля — те, по которым
 // ходят запросы; полный набор кладётся рядом, потому что документация HAE

internal/ingest/ingest.go

@@ -14,7 +14,6 @@ import (
 	"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/store"
 )
@@ -48,33 +47,56 @@ type Result struct {
 	RawPath    string
 }
 
-// foldTimeout — сколько отводится свёртке принятой доставки.
+// recordTimeout — сколько отводится записи учёта доставки.
 //
-// Свёртка идёт на контексте, отвязанном от запроса, поэтому собственный
-// дедлайн обязателен: без него зависшая запись держала бы горутину до конца
-// жизни процесса.
-const foldTimeout = 2 * time.Minute
+// Учёт ведётся на контексте, переживающем обрыв соединения (см. Accept),
+// поэтому собственный дедлайн обязателен: без него отказ базы держал бы
+// обработчик неограниченно.
+const recordTimeout = 10 * time.Second
 
-// Service принимает пакеты: сохраняет тело в архив, учитывает доставку и
-// запускает её свёртку.
+// Notify — «есть работа»: сигнал тому, кто сворачивает принятое.
+//
+// Функцией, а не интерфейсом: сигнал ничего не несёт и ничего не возвращает,
+// а приёму незачем знать, кто именно свернёт доставку.
+type Notify func()
+
+// Service принимает пакеты: сохраняет тело в архив, учитывает доставку и будит
+// свёртку.
+//
+// Сворачивать сам он не умеет намеренно. Свёртка широкой доставки идёт
+// секундами, а `WriteTimeout` в Go ставится до вызова обработчика — то есть
+// синхронная свёртка тратила бы бюджет ответа и обрывала бы соединение молча,
+// с записью `status_code=200` в журнале доступа.
 type Service struct {
-	arch  *archive.Archive
-	store *store.Store
-	fold  *fold.Service
-	log   *slog.Logger
+	arch   *archive.Archive
+	store  *store.Store
+	notify Notify
+	log    *slog.Logger
 }
 
 // New собирает use-case приёма.
-func New(arch *archive.Archive, st *store.Store, f *fold.Service, log *slog.Logger) *Service {
-	return &Service{arch: arch, store: st, fold: f, log: log.With("capability", "ingest")}
+//
+// Нулевой notify означает «о свёртке заботится вызывающий» и приводится к
+// пустой функции здесь же, один раз: проверка на nil в месте вызова рано или
+// поздно окажется забытой, а паника там наступила бы ПОСЛЕ того, как тело уже
+// записано и доставка учтена, — то есть отправитель получил бы отказ по
+// сохранённой доставке.
+func New(arch *archive.Archive, st *store.Store, notify Notify, log *slog.Logger) *Service {
+	if notify == nil {
+		notify = func() {}
+	}
+	return &Service{arch: arch, store: st, notify: notify, log: log.With("capability", "ingest")}
 }
 
-// Accept принимает тело пакета: проверяет форму, кладёт в сырой архив и
-// заводит запись о доставке.
+// Accept принимает тело пакета: проверяет форму, кладёт в сырой архив, заводит
+// запись о доставке и будит свёртку.
 //
 // Порядок важен: сначала тело оказывается на диске, и только потом появляется
 // учётная запись. Обратный порядок дал бы учтённую доставку без данных.
 //
+// Возврат означает «сохранено и учтено», а не «разобрано»: доставка уезжает в
+// очередь свёртки статусом `pending`, и её исход появится позже.
+//
 // Это единственный логирующий чекпоинт приёма — транспорт исход не логирует.
 func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, error) {
 	if err := checkEnvelope(body); err != nil {
@@ -94,9 +116,14 @@ func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, e
 	// к часам. Источник обязан быть один: у тела, лежащего в архиве без учётной
 	// записи, метку восстанавливают из ULID, и два разных источника разошлись бы
 	// на границе секунды — а от порядка журнала зависит наследование слоя.
+	// Ошибка здесь означает, что наш же генератор выдал неразбираемый
+	// идентификатор. Второго источника времени тут быть не может — он разошёлся
+	// бы с меткой, которую пересборка восстанавливает из ULID; поэтому отказ, а
+	// не подмена. Тело на диск ещё не легло, так что доставка не теряется.
 	receivedAt, err := ident.TimeOf(res.DeliveryID)
 	if err != nil {
-		receivedAt = store.Now()
+		s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID)
+		return Result{}, fmt.Errorf("метка приёма из идентификатора: %w", err)
 	}
 
 	rawPath, err := s.arch.Write(res.DeliveryID, receivedAt, body)
@@ -106,7 +133,15 @@ func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, e
 	}
 	res.RawPath = rawPath
 
-	err = s.store.CreateDelivery(ctx, store.Delivery{
+	// Учёт ведётся на контексте, ПЕРЕЖИВАЮЩЕМ обрыв соединения. Тело к этому
+	// моменту уже на диске (arch.Write контекста не берёт), и отказ вставки
+	// из-за ушедшего клиента оставил бы тело сиротой: доставки в журнале нет,
+	// а вернуть её может только пересборка с ручной подменой базы. Проверка
+	// формы выше остаётся отменяемой — там отмена уместна.
+	recordCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), recordTimeout)
+	defer cancel()
+
+	err = s.store.CreateDelivery(recordCtx, store.Delivery{
 		ID:             res.DeliveryID,
 		ReceivedAt:     receivedAt,
 		Headers:        encodeHeaders(meta.Headers),
@@ -121,8 +156,8 @@ func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, e
 		ParseStatus:    store.ParsePending,
 	})
 	if err != nil {
-		// Тело уже на диске — данные не потеряны, но учёта нет. Разбор архива
-		// на следующем шаге проекта такую доставку подберёт.
+		// Тело уже на диске — данные не потеряны, но учёта нет. Такое тело
+		// подберёт пересборка (`healthlog reindex`), заведя запись заново.
 		s.log.ErrorContext(ctx, "delivery failed", "error", err, "delivery_id", res.DeliveryID, "raw_path", rawPath)
 		return Result{}, fmt.Errorf("record delivery: %w", err)
 	}
@@ -131,24 +166,37 @@ func (s *Service) Accept(ctx context.Context, body []byte, meta Meta) (Result, e
 		"delivery_id", res.DeliveryID,
 		"bytes", res.Bytes,
 		"raw_path", rawPath,
-		"automation_name", meta.AutomationName,
-		"aggregation", meta.Aggregation,
-		"period", meta.Period)
+		"automation_name", clip(meta.AutomationName),
+		"aggregation", clip(meta.Aggregation),
+		"period", clip(meta.Period))
 
-	// Свёртка идёт после того, как доставка учтена, и на контексте, ОТВЯЗАННОМ
-	// от запроса: обрыв соединения клиентом или прокси на середине оставил бы
-	// часть объектов записанной, а доставку — со статусом, по которому её
-	// никто не подберёт. Исход свёртки на код ответа не влияет — сохранили
-	// значит приняли.
-	foldCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), foldTimeout)
-	defer cancel()
-	// Ошибку не возвращаем: она уже записана в лог и в parse_status свёрткой,
-	// а доставка принята.
-	_, _ = s.fold.Fold(foldCtx, res.DeliveryID)
+	// Сигнал идёт последним — после того, как строка учёта закоммичена: иначе
+	// воркер мог бы проснуться раньше, чем увидит доставку, и потратить проход
+	// впустую. Потеря сигнала отказом не является: доставка числится `pending`,
+	// и её подберёт следующий сигнал, тик воркера или старт сервиса.
+	s.notify()
 
 	return res, nil
 }
 
+// maxAttrLen — сколько байт значения заголовка попадает в лог.
+//
+// Заголовки контролирует отправитель целиком, а `MaxHeaderBytes` у Go — мегабайт
+// на запрос: без границы одна доставка выдавливает из ротации логов всю недавнюю
+// историю, включая записи, по которым эту же доставку потом разыскивают. Та же
+// граница по той же причине стоит на именах метрик и секций.
+const maxAttrLen = 128
+
+// clip обрезает значение, пришедшее от отправителя, до пригодного для лога.
+func clip(s string) string {
+	if len(s) <= maxAttrLen {
+		return s
+	}
+	// Обрезка названа в самом значении: молча укороченное имя автоматизации
+	// выглядит как другое имя.
+	return s[:maxAttrLen] + "…(обрезано)"
+}
+
 // encodeHeaders сериализует заголовки для хранения. Ключи json.Marshal
 // сортирует сам, поэтому запись стабильна и её удобно сравнивать между
 // доставками. Сбой сериализации не должен ронять приём: заголовки —

internal/ingest/ingest_test.go

@@ -12,7 +12,6 @@ import (
 	"testing"
 
 	"git.vakhrushev.me/av/healthlog/internal/archive"
-	"git.vakhrushev.me/av/healthlog/internal/fold"
 	"git.vakhrushev.me/av/healthlog/internal/ingest"
 	"git.vakhrushev.me/av/healthlog/internal/store"
 )
@@ -123,32 +122,9 @@ func TestAcceptRejectsMalformed(t *testing.T) {
 	}
 }
 
-// Разбор не влияет на исход приёма: сохранили — значит приняли. Непонятое
-// содержимое даёт принятую доставку с parse_status=failed, а не отказ.
-func TestAcceptНепонятоеСодержимоеПринимается(t *testing.T) {
-	svc, _, st := newService(t)
-	ctx := context.Background()
-
-	// Метрика есть, но слой определить нечем: плотных метрик нет, заголовок
-	// ничего не означает, наследовать не от чего.
-	body := []byte(`{"data":{"metrics":[{"name":"m","units":"u","data":[` +
-		`{"date":"2026-07-31 12:00:00 +0300","qty":1}]}]}}`)
-
-	if _, err := svc.Accept(ctx, body, ingest.Meta{Aggregation: "Default"}); err != nil {
-		t.Fatalf("Accept отверг доставку из-за разбора: %v", err)
-	}
-
-	d, err := st.LastDelivery(ctx)
-	if err != nil {
-		t.Fatalf("LastDelivery: %v", err)
-	}
-	if d.ParseStatus != store.ParseFailed {
-		t.Errorf("parse_status = %q, ожидался %q", d.ParseStatus, store.ParseFailed)
-	}
-}
-
-// Разобранная доставка отмечается разобранной, и точки доезжают до объектов.
-func TestAcceptРазобраннаяДоставкаОтмечена(t *testing.T) {
+// Ответ отдаётся ДО свёртки: принятая доставка ждёт разбора в очереди, а не
+// приезжает разобранной. Это смена контракта, и она проверяется явно.
+func TestAcceptОставляетДоставкуВОчереди(t *testing.T) {
 	svc, _, st := newService(t)
 	ctx := context.Background()
 
@@ -165,32 +141,106 @@ func TestAcceptРазобраннаяДоставкаОтмечена(t *testing
 	if err != nil {
 		t.Fatalf("LastDelivery: %v", err)
 	}
-	if d.ParseStatus != store.ParseDone {
-		t.Fatalf("parse_status = %q, ожидался %q", d.ParseStatus, store.ParseDone)
-	}
-	if d.Points == 0 {
-		t.Error("точек 0: разбор не дошёл до учёта")
+	if d.ParseStatus != store.ParsePending {
+		t.Errorf("parse_status = %q, ожидался %q", d.ParseStatus, store.ParsePending)
 	}
 
 	n, err := st.CountBuckets(ctx)
 	if err != nil {
 		t.Fatalf("CountBuckets: %v", err)
 	}
-	if n == 0 {
-		t.Error("объектов 0: точки не доехали до хранилища")
+	if n != 0 {
+		t.Errorf("объектов %d: свёртка произошла внутри приёма", n)
 	}
 }
 
-// Свёртка идёт на контексте, отвязанном от запроса (context.WithoutCancel в
-// Accept), чтобы обрыв соединения не оставил часть объектов записанной.
-// Автотестом это не покрыто: отмену надо подать РОВНО между учётом доставки и
-// свёрткой, а такого шва снаружи нет, и заводить его ради теста дороже, чем
-// проверять глазами. Атомарность самой записи проверена в store
-// (TestMergePointsОтменаНеОставляетПоловины).
-
-func newService(t *testing.T) (*ingest.Service, *archive.Archive, *store.Store) {
-	t.Helper()
+// Сигнал уходит после того, как доставка учтена: воркер, разбуженный раньше,
+// потратил бы проход впустую.
+func TestAcceptБудитСвёрткуПослеУчёта(t *testing.T) {
 	dir := t.TempDir()
+	st, arch := newDeps(t, dir)
+
+	var seen int64
+	notify := func() {
+		n, err := st.CountDeliveries(context.Background())
+		if err != nil {
+			t.Errorf("CountDeliveries: %v", err)
+		}
+		seen = n
+	}
+	svc := ingest.New(arch, st, notify, slog.New(slog.DiscardHandler))
+
+	if _, err := svc.Accept(context.Background(), []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}); err != nil {
+		t.Fatalf("Accept: %v", err)
+	}
+	if seen != 1 {
+		t.Errorf("на момент сигнала доставок в учёте %d, ожидалась 1", seen)
+	}
+}
+
+// Нулевой сигнал — законный вход (свёрткой заведует вызывающий), и приём от
+// него не падает. Паника здесь наступила бы ПОСЛЕ записи тела и учёта, то есть
+// отправитель получил бы отказ по сохранённой доставке.
+func TestAcceptБезСигналаНеПадает(t *testing.T) {
+	dir := t.TempDir()
+	st, arch := newDeps(t, dir)
+	svc := ingest.New(arch, st, nil, slog.New(slog.DiscardHandler))
+
+	if _, err := svc.Accept(context.Background(), []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}); err != nil {
+		t.Fatalf("Accept: %v", err)
+	}
+}
+
+// Обрыв соединения после записи тела не должен оставлять тело без учёта:
+// доставка, не попавшая в журнал, восстанавливается только пересборкой с
+// ручной подменой базы.
+func TestAcceptУчитываетДоставкуПослеОбрываСоединения(t *testing.T) {
+	svc, _, st := newService(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	res, err := svc.Accept(ctx, []byte(`{"data":{"metrics":[]}}`), ingest.Meta{})
+	if err != nil {
+		t.Fatalf("Accept на отменённом контексте: %v", err)
+	}
+
+	status, err := st.DeliveryStatus(context.Background(), res.DeliveryID)
+	if err != nil {
+		t.Fatalf("DeliveryStatus: %v", err)
+	}
+	if status != store.ParsePending {
+		t.Errorf("parse_status = %q, ожидался %q", status, store.ParsePending)
+	}
+}
+
+// Учёта нет, а тело есть: приём кладёт тело на диск раньше строки в базе, и
+// отказ на вставке оставляет тело в архиве. Такое тело подберёт пересборка.
+func TestAcceptПриОтказеУчётаОставляетТелоВАрхиве(t *testing.T) {
+	dir := t.TempDir()
+	st, arch := newDeps(t, dir)
+	svc := ingest.New(arch, st, nil, slog.New(slog.DiscardHandler))
+
+	// База закрыта — учесть доставку нечем.
+	if err := st.Close(); err != nil {
+		t.Fatalf("закрытие базы: %v", err)
+	}
+
+	if _, err := svc.Accept(context.Background(), []byte(`{"data":{"metrics":[]}}`), ingest.Meta{}); err == nil {
+		t.Fatal("приём не заметил, что доставка не учтена")
+	}
+
+	entries, err := filepath.Glob(filepath.Join(arch.Root(), "*", "*", "*", "*.json.gz"))
+	if err != nil {
+		t.Fatalf("обход архива: %v", err)
+	}
+	if len(entries) != 1 {
+		t.Errorf("тел в архиве %d, ожидалось 1: тело потеряно вместе с учётом", len(entries))
+	}
+}
+
+func newDeps(t *testing.T, dir string) (*store.Store, *archive.Archive) {
+	t.Helper()
 
 	st, err := store.Open(filepath.Join(dir, "healthlog.db"))
 	if err != nil {
@@ -202,7 +252,12 @@ func newService(t *testing.T) (*ingest.Service, *archive.Archive, *store.Store)
 	if err != nil {
 		t.Fatalf("archive.New: %v", err)
 	}
+	return st, arch
+}
 
-	log := slog.New(slog.DiscardHandler)
-	return ingest.New(arch, st, fold.New(arch, st, 0, log), log), arch, st
+func newService(t *testing.T) (*ingest.Service, *archive.Archive, *store.Store) {
+	t.Helper()
+
+	st, arch := newDeps(t, t.TempDir())
+	return ingest.New(arch, st, nil, slog.New(slog.DiscardHandler)), arch, st
 }

internal/replay/classify_test.go

@@ -0,0 +1,67 @@
+package replay
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"git.vakhrushev.me/av/healthlog/internal/fold"
+	"git.vakhrushev.me/av/healthlog/internal/hae"
+	"git.vakhrushev.me/av/healthlog/internal/store"
+)
+
+// Классификация исхода — та половина, которую пересборка и фоновый воркер
+// обязаны делить. Проверяется перебором классов, без базы и без архива: второй
+// классификатор разошёлся бы с первым молча, а по счётчику `partial`
+// принимается решение о судьбе тела в архиве.
+func TestClassifyРазводитИсходыПоКлассам(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		err  error
+		want Outcome
+	}{
+		{"успех", nil, Outcome{Folded: 1}},
+		{"слой не выведен", hae.ErrLayerUnknown, Outcome{FailedLayer: 1}},
+		{"слой не выведен, обёрнут", fmt.Errorf("свёртка: %w", hae.ErrLayerUnknown), Outcome{FailedLayer: 1}},
+		{"содержимое не разбирается", hae.ErrMalformed, Outcome{FailedMalformed: 1}},
+		{"база занята", store.ErrBusy, Outcome{Deferred: 1}},
+		{"база занята, обёрнута", fmt.Errorf("слияние: %w", store.ErrBusy), Outcome{Deferred: 1}},
+		{"работу прекратили снаружи", context.Canceled, Outcome{Deferred: 1}},
+		// Дедлайн — свойство доставки, а не обстоятельств: она не уложится в
+		// бюджет и в следующий раз, а повтор безнадёжного останавливает очередь.
+		{"свёртка не уложилась в бюджет", context.DeadlineExceeded, Outcome{FailedOther: 1}},
+		{"прочее", errors.New("диск отвалился"), Outcome{FailedOther: 1}},
+		// Паника — дефект нашего кода, а не обстоятельство: доставка выводится
+		// из очереди, тело ждёт пересборки.
+		{"свёртка паниковала", fold.ErrPanicked, Outcome{FailedOther: 1}},
+	}
+
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			t.Parallel()
+
+			if got := classify(c.err); got != c.want {
+				t.Errorf("Classify(%v) = %+v, ожидалось %+v", c.err, got, c.want)
+			}
+		})
+	}
+}
+
+// Накопление — сложение по классам: у пересборки и у воркера один набор имён
+// для одних исходов.
+func TestOutcomeAddСкладываетПоКлассам(t *testing.T) {
+	t.Parallel()
+
+	var total Outcome
+	total.Add(Outcome{Folded: 1, Partial: 1})
+	total.Add(Outcome{Folded: 1, Incomparable: 2})
+	total.Add(Outcome{Deferred: 1})
+
+	want := Outcome{Folded: 2, Deferred: 1, Partial: 1, Incomparable: 2}
+	if total != want {
+		t.Errorf("сумма %+v, ожидалась %+v", total, want)
+	}
+}

internal/replay/player.go

@@ -0,0 +1,110 @@
+package replay
+
+import (
+	"context"
+	"errors"
+
+	"git.vakhrushev.me/av/healthlog/internal/fold"
+	"git.vakhrushev.me/av/healthlog/internal/hae"
+	"git.vakhrushev.me/av/healthlog/internal/store"
+)
+
+// Outcome — исход свёртки: одной доставки или их последовательности.
+//
+// Классы разведены потому, что читаются по-разному. FailedLayer — штатный исход
+// (слой не выводится, таких тел в журнале заведомо есть), FailedMalformed —
+// содержимое не разбирается, Deferred — работа не сделана по обстоятельствам,
+// и доставка осталась в очереди. Только FailedOther означает, что что-то не так
+// с самой свёрткой. Один общий счётчик отправлял бы человека искать дефект там,
+// где его нет.
+type Outcome struct {
+	Folded          int
+	FailedLayer     int
+	FailedMalformed int
+	// Deferred — доставка осталась `pending`: отмена или занятость базы. Не
+	// отказ доставки, а несделанная работа; её подберёт следующий проход.
+	Deferred    int
+	FailedOther int
+	// Partial — доставок, в теле которых остались непокрытые разбором секции.
+	// Не отклонение, а половина потока; названо потому, что именно эти тела
+	// ретеншену трогать нельзя.
+	Partial int
+	// Incomparable — столкновений с несравнимыми наборами полей. На живом потоке
+	// их не было ни разу, и на этом стоит отказ от объединения полей.
+	Incomparable int
+}
+
+// Add накапливает исход одной доставки в общий.
+func (o *Outcome) Add(other Outcome) {
+	o.Folded += other.Folded
+	o.FailedLayer += other.FailedLayer
+	o.FailedMalformed += other.FailedMalformed
+	o.Deferred += other.Deferred
+	o.FailedOther += other.FailedOther
+	o.Partial += other.Partial
+	o.Incomparable += other.Incomparable
+}
+
+// classify раскладывает ошибку свёртки по классам исхода.
+//
+// Чистая функция, и это не украшение: она и есть та половина, которую задача
+// требовала не дублировать между пересборкой и фоновым воркером, — а
+// проверяется она перебором классов, без базы и без архива.
+//
+// Неэкспортируемая намеренно: её результат содержит поля `Partial` и
+// `Incomparable`, которые дописывает только Play, — вторая публичная дверь
+// молча занижала бы именно тот счётчик, по которому принимается решение о
+// судьбе тела в архиве.
+func classify(err error) Outcome {
+	var out Outcome
+	switch {
+	case err == nil:
+		out.Folded++
+	case store.Transient(err):
+		// Статус доставки свёртка в этих случаях не трогает: она осталась
+		// `pending` и будет свёрнута снова. Правило одно на обоих — то, по
+		// которому свёртка решает не писать исход.
+		out.Deferred++
+	case errors.Is(err, hae.ErrLayerUnknown):
+		out.FailedLayer++
+	case errors.Is(err, hae.ErrMalformed):
+		out.FailedMalformed++
+	default:
+		out.FailedOther++
+	}
+	return out
+}
+
+// Player сворачивает доставку по идентификатору и классифицирует исход.
+//
+// Общий и для пересборки журнала, и для фонового воркера приёма — второй
+// классификатор разошёлся бы с первым молча, а по одному из его счётчиков
+// (`Partial`) принимается решение о судьбе тела в архиве.
+type Player struct {
+	Fold *fold.Service
+}
+
+// Play сворачивает одну доставку и возвращает её исход.
+//
+// Классифицируется ТОЛЬКО ошибка свёртки: на контекст Play не смотрит, и это
+// существенно. У двух вызывающих отменённый контекст означает противоположное —
+// у пересборки в свёртку уходит тот же отменяемый контекст («нас остановили»),
+// у воркера отвязанный от остановки, с собственным дедлайном («доставка не
+// уложилась в бюджет»). Решение «работу прекратили снаружи» принимает цикл,
+// каждый по своему контексту.
+func (p Player) Play(ctx context.Context, deliveryID string) (Outcome, error) {
+	st, err := p.Fold.Fold(ctx, deliveryID)
+	out := classify(err)
+
+	if err == nil {
+		// Счётчики читаются только у успешной свёртки: при ошибке поля Stats
+		// заполнены частично (Uncovered у отказавшего разбора всегда пуст, хотя
+		// в базу список записан) — и Partial молча занижался бы. А по нему
+		// принимается решение о ретеншене тел.
+		if len(st.Uncovered) > 0 {
+			out.Partial++
+		}
+		out.Incomparable += st.Incomparable
+	}
+	return out, err
+}

internal/replay/replay.go

@@ -24,7 +24,6 @@ import (
 
 	"git.vakhrushev.me/av/healthlog/internal/archive"
 	"git.vakhrushev.me/av/healthlog/internal/fold"
-	"git.vakhrushev.me/av/healthlog/internal/hae"
 	"git.vakhrushev.me/av/healthlog/internal/ident"
 	"git.vakhrushev.me/av/healthlog/internal/store"
 )
@@ -70,24 +69,11 @@ type Report struct {
 	// Orphans — строк учёта, у которых тела в архиве нет. Станет штатным, когда
 	// появится ретеншен архива.
 	Orphans int
-	// Folded — сколько доставок свернулось.
-	Folded int
-	// Отказы разведены по классам, потому что читаются они по-разному.
-	// FailedLayer — слой не выводится: штатный исход, таких доставок в журнале
-	// заведомо есть. FailedMalformed — содержимое не разбирается. FailedOther —
-	// всё прочее (тело не читается, отказ базы); только оно означает, что с
-	// пересборкой что-то не так. Один общий счётчик отправлял бы человека
-	// искать дефект там, где его нет.
-	FailedLayer     int
-	FailedMalformed int
-	FailedOther     int
-	// Partial — доставок, в теле которых остались непокрытые разбором секции.
-	// Не отклонение, а половина потока; названо потому, что именно эти тела
-	// ретеншену трогать нельзя.
-	Partial int
-	// Incomparable — столкновений с несравнимыми наборами полей. На живом потоке
-	// их не было ни разу, и на этом стоит отказ от объединения полей.
-	Incomparable int
+
+	// Outcome — счётчики свёртки, те же самые, что считает фоновый воркер
+	// приёма. Встроены, а не продублированы именами: два набора имён для одних
+	// исходов разошлись бы при первой же правке классификации.
+	Outcome
 
 	Buckets     int64
 	Fingerprint string
@@ -148,6 +134,7 @@ func Run(ctx context.Context, o Options) (Report, error) {
 		}
 	}
 
+	player := Player{Fold: o.Fold}
 	for i, d := range journal {
 		if ctx.Err() != nil {
 			rep.Canceled = true
@@ -156,36 +143,17 @@ func Run(ctx context.Context, o Options) (Report, error) {
 		if err := o.Target.CreateDelivery(ctx, d); err != nil {
 			return stopOr(rep, err)
 		}
-		st, err := o.Fold.Fold(ctx, d.ID)
-		switch {
-		case err == nil:
-			rep.Folded++
-		case ctx.Err() != nil:
-			// Отмена, застигшая свёртку, — не отказ доставки: считать её отказом
-			// значило бы обвинить разбор в том, чего он не делал, и отправить
-			// человека искать дефект по логу.
+		out, _ := player.Play(ctx, d.ID)
+		// Отмена, застигшая свёртку, — не отказ доставки: считать её отказом
+		// значило бы обвинить разбор в том, чего он не делал, и отправить
+		// человека искать дефект по логу. Решение принимает цикл по СВОЕМУ
+		// контексту — тому же, на котором шла свёртка; классификатор о нём не
+		// знает намеренно, у воркера тот же признак означает другое.
+		if ctx.Err() != nil {
 			rep.Canceled = true
 			return rep, nil
-		case errors.Is(err, hae.ErrLayerUnknown):
-			// Штатный исход, уже записанный свёрткой в лог и в parse_status:
-			// журнал заведомо содержит тела без плотных метрик. Останов на
-			// первом лишил бы пересборки все остальные.
-			rep.FailedLayer++
-		case errors.Is(err, hae.ErrMalformed):
-			rep.FailedMalformed++
-		default:
-			rep.FailedOther++
-		}
-		if err == nil {
-			// Счётчики читаются только у успешной свёртки: при ошибке поля Stats
-			// заполнены частично (Uncovered у отказавшего разбора всегда пуст,
-			// хотя в базу список записан) — и Partial молча занижался бы. А по
-			// нему принимается решение о ретеншене тел.
-			if len(st.Uncovered) > 0 {
-				rep.Partial++
-			}
-			rep.Incomparable += st.Incomparable
 		}
+		rep.Add(out)
 		if o.Progress != nil {
 			o.Progress(i+1, len(journal))
 		}
@@ -218,6 +186,7 @@ func Run(ctx context.Context, o Options) (Report, error) {
 		"folded", rep.Folded,
 		"failed_layer", rep.FailedLayer,
 		"failed_malformed", rep.FailedMalformed,
+		"deferred", rep.Deferred,
 		"failed_other", rep.FailedOther,
 		"partial", rep.Partial,
 		"incomparable", rep.Incomparable,

internal/replay/worker.go

@@ -0,0 +1,232 @@
+package replay
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"git.vakhrushev.me/av/healthlog/internal/fold"
+	"git.vakhrushev.me/av/healthlog/internal/store"
+)
+
+// batchSize — сколько неразобранных доставок берётся одним запросом.
+//
+// Не ради страниц, а ради памяти: задолженность после миграции, переводящей
+// строки в `pending`, равна всему архиву, и материализовать её целиком незачем —
+// проход всё равно идёт по одной.
+const batchSize = 256
+
+// foldTimeout — сколько отводится свёртке одной доставки.
+//
+// Свёртка идёт на контексте, отвязанном от остановки, поэтому собственный
+// дедлайн обязателен: без него зависшая запись держала бы единственного воркера
+// до конца жизни процесса, и очередь перестала бы двигаться вовсе.
+const foldTimeout = 2 * time.Minute
+
+// tickInterval — как часто воркер просыпается сам, без сигнала.
+//
+// Сигнал приносит приём, и для свежей доставки его достаточно. Тик закрывает
+// два случая, которых сигнал не закрывает: доставка, оставшаяся в очереди из-за
+// занятости базы, иначе ждала бы СЛЕДУЮЩЕЙ доставки (а ночью телефон молчит
+// часами), и метка отставания иначе не вычислялась бы вовсе — «работа есть,
+// прогресса нет» было бы неотличимо от здорового пустого потока.
+const tickInterval = time.Minute
+
+// lagThreshold — с какого ожидания доставка считается задержанной.
+//
+// Период быстрого прохода синхронизации: если доставка ждала дольше, чем
+// интервал между доставками, очередь растёт, а не рассасывается.
+const lagThreshold = 5 * time.Minute
+
+// Worker — фоновая свёртка принятых доставок.
+//
+// Очередь — сама таблица: доставка ждёт свёртки в статусе `pending`, а канал
+// несёт только бит «есть работа». Отсюда три свойства, ради которых так и
+// сделано: переполнять нечего, падение процесса очереди не теряет, а подбор
+// неразобранного при старте не является отдельным кодом — это обычный проход.
+type Worker struct {
+	store  *store.Store
+	player Player
+	log    *slog.Logger
+
+	// wake — сигнал «есть работа», ёмкость 1 и неблокирующая отправка. Та же
+	// форма, что у os/signal.Notify: сигнал ничего не несёт, и потерять лишний
+	// не только можно, но и нужно.
+	wake chan struct{}
+
+	// startupDone — первый проход завершён.
+	//
+	// До него метка отставания молчит: задолженность, накопленная ДО старта,
+	// ждала не воркера, а его появления, и сотня одинаковых WARN при первом же
+	// запуске обесценила бы уровень.
+	startupDone bool
+}
+
+// NewWorker собирает воркер над рабочей базой.
+func NewWorker(st *store.Store, f *fold.Service, log *slog.Logger) *Worker {
+	return &Worker{
+		store:  st,
+		player: Player{Fold: f},
+		log:    log.With("capability", "fold-worker"),
+		wake:   make(chan struct{}, 1),
+	}
+}
+
+// Notify будит воркер. Вызывается приёмом после того, как доставка учтена.
+//
+// Потеря сигнала отказом не является: доставка от этого не перестаёт числиться
+// `pending`, и её подберёт следующий сигнал, тик или старт.
+func (w *Worker) Notify() {
+	select {
+	case w.wake <- struct{}{}:
+	default:
+	}
+}
+
+// Run ведёт воркер до отмены контекста.
+//
+// Отмена проверяется МЕЖДУ доставками: свёртка идёт на отвязанном контексте и
+// рваться не должна. Обещания «текущая доставка непременно досворачивается» тут
+// нет — бюджет остановки меньше бюджета свёртки; гарантируется другое: после
+// выхода не существует доставки, которая числится разобранной, а записана
+// наполовину.
+//
+// Первый проход делается сразу, без ожидания сигнала: он и есть подбор
+// неразобранного при старте.
+func (w *Worker) Run(ctx context.Context) {
+	if n, err := w.store.CountPendingDeliveries(ctx); err != nil {
+		if ctx.Err() == nil {
+			w.log.ErrorContext(ctx, "pending backlog not counted", "error", err)
+		}
+	} else if n > 0 {
+		// Размер задолженности — ответ на вопрос «что сервис будет делать
+		// первые минуты после рестарта». Одной строкой и один раз.
+		w.log.InfoContext(ctx, "pending backlog at start", "deliveries", n)
+	}
+
+	ticker := time.NewTicker(tickInterval)
+	defer ticker.Stop()
+
+	for {
+		if _, err := w.Pass(ctx); err != nil && ctx.Err() == nil {
+			// Отказ прохода не убивает цикл: воркер, умерший от временного
+			// отказа базы, остановил бы свёртку до конца жизни процесса, пока
+			// приём продолжал бы отвечать 200.
+			//
+			// Отмена сюда не попадает: штатная остановка не отказ, а ERROR о
+			// ней обесценил бы уровень, по которому вмешиваются.
+			w.log.ErrorContext(ctx, "fold pass failed", "error", err)
+		}
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-w.wake:
+		case <-ticker.C:
+		}
+	}
+}
+
+// Pass делает один проход по очереди и возвращает его исход.
+//
+// Синхронный шов: тесты зовут его напрямую и не ждут по часам. Без него
+// проверки «все свёрнуты», «проход конечен», «метка не сработала на первом
+// проходе» писались бы опросом базы с таймаутом.
+//
+// Курсор строго возрастает, и это нужно не ради страниц, а ради завершимости:
+// доставка, у которой не удалось записать даже исход разбора, остаётся
+// `pending`, и проход без курсора выбирал бы её бесконечно.
+func (w *Worker) Pass(ctx context.Context) (Outcome, error) {
+	var total Outcome
+	var cursor store.PendingDelivery
+	var lag lagged
+
+	for {
+		if ctx.Err() != nil {
+			return total, nil
+		}
+
+		batch, err := w.store.PendingDeliveries(ctx, cursor, batchSize)
+		if err != nil {
+			return total, err
+		}
+		if len(batch) == 0 {
+			// Флаг снимается ТОЛЬКО здесь — у прохода, дошедшего до пустой
+			// выборки. Взведённый на любом выходе (отказ базы, отмена), он
+			// включал бы метку задержки после прохода, который ничего не
+			// свернул, и следующий проход выдал бы WARN на всю задолженность —
+			// ровно тот шквал, против которого метка и подавляется при старте.
+			//
+			// Порядок двух строк существен: на задолженности ПЕРВОГО прохода
+			// метка молчит — та ждала не воркера, а его появления.
+			w.warnLag(ctx, lag)
+			w.startupDone = true
+			return total, nil
+		}
+
+		for _, d := range batch {
+			if ctx.Err() != nil {
+				return total, nil
+			}
+			lag.add(d)
+			total.Add(w.foldOne(ctx, d.ID))
+			cursor = d
+		}
+	}
+}
+
+// lagged копит отставание прохода: сколько доставок ждали свёртки и дольше всех
+// ждала какая.
+//
+// Считается на ВЫБОРКЕ, а не по факту успешной свёртки: иначе застрявшая
+// доставка молчала бы ровно в том состоянии, ради которого метка и заведена.
+type lagged struct {
+	count   int
+	worst   time.Duration
+	worstID string
+}
+
+func (l *lagged) add(d store.PendingDelivery) {
+	waited := store.Now().Sub(d.ReceivedAt)
+	if waited < lagThreshold {
+		return
+	}
+	l.count++
+	if waited > l.worst {
+		l.worst = waited
+		l.worstID = d.ID
+	}
+}
+
+// foldOne сворачивает доставку на контексте, ОТВЯЗАННОМ от остановки.
+//
+// Отмена снаружи не должна превращаться в свойство доставки: свёртка пишет
+// исход на переживающем отмену контексте, и оборванная на середине пометила бы
+// доставку так, что воркер её больше не подберёт. Собственный дедлайн при этом
+// остаётся и означает именно отказ доставки.
+func (w *Worker) foldOne(ctx context.Context, deliveryID string) Outcome {
+	foldCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), foldTimeout)
+	defer cancel()
+
+	// Ошибку не возвращаем: она уже записана свёрткой в лог и в parse_status,
+	// а отказ одной доставки прохода не прекращает. Паника тоже: её
+	// перехватывает сама свёртка — там же, где живёт единственный писатель
+	// исхода разбора.
+	out, _ := w.player.Play(foldCtx, deliveryID)
+	return out
+}
+
+// warnLag называет отставание одной строкой на проход.
+//
+// Одной, а не по строке на доставку: задолженность в сотню тел давала бы сотню
+// одинаковых WARN каждую минуту, и уровень, по которому вмешиваются, перестал
+// бы что-либо значить.
+func (w *Worker) warnLag(ctx context.Context, l lagged) {
+	if !w.startupDone || l.count == 0 {
+		return
+	}
+	w.log.WarnContext(ctx, "deliveries waited for fold",
+		"deliveries", l.count,
+		"worst_delivery_id", l.worstID,
+		"worst_waited_sec", int64(l.worst.Seconds()))
+}

internal/replay/worker_test.go

@@ -0,0 +1,373 @@
+package replay_test
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"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"
+)
+
+// newWorker собирает воркер над свежей базой и архивом, отдавая заодно то, чем
+// проверяют его следы в логе.
+func newWorker(t *testing.T, dir string) (*replay.Worker, *store.Store, *archive.Archive, *logSink) {
+	t.Helper()
+
+	arch := openArchive(t, filepath.Join(dir, "raw"))
+	st := openStore(t, filepath.Join(dir, "live.db"))
+	sink := newLogSink()
+	log := slog.New(slog.NewJSONHandler(sink, &slog.HandlerOptions{Level: slog.LevelDebug}))
+	w := replay.NewWorker(st, fold.New(arch, st, 0, log), log)
+	return w, st, arch, sink
+}
+
+// logSink собирает записи лога, чтобы проверять их без гонок и без ожиданий по
+// часам: тест синхронизируется появлением строки, а не сном.
+type logSink struct {
+	mu    sync.Mutex
+	lines []string
+	watch map[string]*watcher
+}
+
+// watcher ждёт n-го появления строки.
+type watcher struct {
+	left int
+	ch   chan struct{}
+}
+
+func newLogSink() *logSink { return &logSink{watch: map[string]*watcher{}} }
+
+func (s *logSink) Write(p []byte) (int, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	line := string(p)
+	s.lines = append(s.lines, line)
+	if w, ok := s.watch[msgOf(line)]; ok {
+		w.left--
+		if w.left == 0 {
+			close(w.ch)
+			delete(s.watch, msgOf(line))
+		}
+	}
+	return len(p), nil
+}
+
+// expect регистрирует ожидание n-го появления строки до того, как она может
+// появиться: синхронизация идёт событием, а не сном.
+func (s *logSink) expect(msg string, n int) <-chan struct{} {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	w := &watcher{left: n, ch: make(chan struct{})}
+	s.watch[msg] = w
+	return w.ch
+}
+
+func msgOf(line string) string {
+	var rec struct {
+		Msg string `json:"msg"`
+	}
+	if err := json.Unmarshal([]byte(line), &rec); err != nil {
+		return ""
+	}
+	return rec.Msg
+}
+
+// count считает записи с данным msg.
+func (s *logSink) count(msg string) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	n := 0
+	for _, line := range s.lines {
+		if msgOf(line) == msg {
+			n++
+		}
+	}
+	return n
+}
+
+func (s *logSink) dump() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return strings.Join(s.lines, "")
+}
+
+// Подбор неразобранного — обычный проход воркера, а не отдельный режим: после
+// миграции 00005 неразобранными числятся все доставки архива, и подобрать их
+// сегодня может только пересборка с ручной подменой базы.
+func TestПроходПодбираетЗадолженность(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	w, st, arch, _ := newWorker(t, dir)
+	ctx := context.Background()
+
+	items := journal(t, "minute.json", "hour.json", "raw.json")
+	for _, it := range items {
+		writeBody(t, arch, st, it, fixture(t, it.fixture))
+	}
+
+	out, err := w.Pass(ctx)
+	if err != nil {
+		t.Fatalf("проход: %v", err)
+	}
+	if out.Folded != len(items) {
+		t.Fatalf("свёрнуто %d из %d: %+v", out.Folded, len(items), out)
+	}
+
+	n, err := st.CountPendingDeliveries(ctx)
+	if err != nil {
+		t.Fatalf("CountPendingDeliveries: %v", err)
+	}
+	if n != 0 {
+		t.Errorf("неразобранными остались %d доставок", n)
+	}
+
+	buckets, err := st.CountBuckets(ctx)
+	if err != nil {
+		t.Fatalf("CountBuckets: %v", err)
+	}
+	if buckets == 0 {
+		t.Error("объектов 0: точки не доехали до хранилища")
+	}
+}
+
+// Порядок задаётся ЖУРНАЛОМ, а не порядком, в котором доставки попали в учёт.
+// Проверяется наблюдаемым следствием: доставка без плотных метрик наследует
+// слой предшествующей ей по `(received_at, id)`.
+//
+// Учёт заполняется в обратном хронологии порядке — так выглядит гонка двух
+// конкурентных приёмов, где поздняя доставка закоммитила строку первой.
+func TestПроходИдётВПорядкеЖурналаАНеВставки(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	w, st, arch, _ := newWorker(t, dir)
+	ctx := context.Background()
+
+	base := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC)
+	minute := item{id: ident.NewID(), at: base.Add(1 * time.Second), automationID: "a", aggregation: "Default", fixture: "minute.json"}
+	sleep := item{id: ident.NewID(), at: base.Add(2 * time.Second), automationID: "a", aggregation: "Default", fixture: "sparse_sleep.json"}
+
+	// Сначала учитывается ПОЗДНЯЯ доставка.
+	writeBody(t, arch, st, sleep, fixture(t, sleep.fixture))
+	writeBody(t, arch, st, minute, fixture(t, minute.fixture))
+
+	out, err := w.Pass(ctx)
+	if err != nil {
+		t.Fatalf("проход: %v", err)
+	}
+	if out.FailedLayer != 0 {
+		t.Fatalf("слой не вывелся у %d доставок: проход пошёл в порядке вставки", out.FailedLayer)
+	}
+
+	// Предшественник — минутная доставка, значит эпизоды сна легли в minute.
+	hours, err := st.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()
+	w, st, arch, _ := newWorker(t, dir)
+	ctx := context.Background()
+
+	items := journal(t, "minute.json", "hour.json")
+	for _, it := range items {
+		writeBody(t, arch, st, it, fixture(t, it.fixture))
+	}
+	// У первой доставки тела больше нет — свернуть её нечем.
+	if err := os.Remove(filepath.Join(arch.Root(), "2026", "08", "01", items[0].id+".json.gz")); err != nil {
+		t.Fatalf("удаление тела: %v", err)
+	}
+
+	done := make(chan replay.Outcome, 1)
+	go func() {
+		out, err := w.Pass(ctx)
+		if err != nil {
+			t.Errorf("проход: %v", err)
+		}
+		done <- out
+	}()
+
+	var out replay.Outcome
+	select {
+	case out = <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("проход не завершился: курсор не двигается")
+	}
+	if out.Folded != 1 || out.FailedOther != 1 {
+		t.Fatalf("исход прохода %+v: ожидались одна свёрнутая и одна отказавшая", out)
+	}
+
+	// Отказавшая доставка выбыла из очереди — иначе следующий проход брал бы её
+	// снова и снова.
+	n, err := st.CountPendingDeliveries(ctx)
+	if err != nil {
+		t.Fatalf("CountPendingDeliveries: %v", err)
+	}
+	if n != 0 {
+		t.Errorf("неразобранными числятся %d доставок, ожидалось 0", n)
+	}
+}
+
+// Метка задержки молчит на задолженности первого прохода и говорит после него:
+// доставки, накопленные до старта, ждали не воркера, а его появления.
+func TestМеткаЗадержкиВключаетсяПослеПервогоПрохода(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	w, st, arch, sink := newWorker(t, dir)
+	ctx := context.Background()
+
+	old := item{
+		id: ident.NewID(), at: store.Now().Add(-time.Hour),
+		automationID: "a", aggregation: "Minutes", fixture: "minute.json",
+	}
+	writeBody(t, arch, st, old, fixture(t, old.fixture))
+
+	if _, err := w.Pass(ctx); err != nil {
+		t.Fatalf("первый проход: %v", err)
+	}
+	if n := sink.count("deliveries waited for fold"); n != 0 {
+		t.Errorf("на задолженности первого прохода %d предупреждений о задержке:\n%s", n, sink.dump())
+	}
+
+	late := item{
+		id: ident.NewID(), at: store.Now().Add(-time.Hour),
+		automationID: "a", aggregation: "Minutes", fixture: "hour.json",
+	}
+	writeBody(t, arch, st, late, fixture(t, late.fixture))
+
+	if _, err := w.Pass(ctx); err != nil {
+		t.Fatalf("второй проход: %v", err)
+	}
+	if n := sink.count("deliveries waited for fold"); n != 1 {
+		t.Errorf("предупреждений о задержке %d, ожидалось 1:\n%s", n, sink.dump())
+	}
+}
+
+// Отмена контекста завершает цикл — без ожиданий по часам: синхронизация идёт
+// возвратом Run, а не сном.
+func TestRunЗавершаетсяПоОтмене(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	w, _, _, sink := newWorker(t, dir)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		w.Run(ctx)
+	}()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatalf("Run не вышел по отмене:\n%s", sink.dump())
+	}
+}
+
+// Задолженность при старте называется одной строкой: это ответ на вопрос «что
+// сервис будет делать первые минуты после рестарта».
+func TestRunНазываетЗадолженностьПриСтарте(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	w, st, arch, sink := newWorker(t, dir)
+
+	items := journal(t, "minute.json", "hour.json")
+	for _, it := range items {
+		writeBody(t, arch, st, it, fixture(t, it.fixture))
+	}
+
+	// Ожидание регистрируется ДО запуска: тест синхронизируется появлением
+	// строки, а не сном.
+	said := sink.expect("pending backlog at start", 1)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		w.Run(ctx)
+	}()
+
+	select {
+	case <-said:
+	case <-time.After(30 * time.Second):
+		t.Fatalf("строки о задолженности нет:\n%s", sink.dump())
+	}
+	cancel()
+	<-done
+}
+
+// Notify не блокирует и не копит: сигнал ничего не несёт, и лишний теряется
+// намеренно.
+func TestNotifyНеБлокирует(t *testing.T) {
+	t.Parallel()
+
+	w, _, _, _ := newWorker(t, t.TempDir())
+	for range 100 {
+		w.Notify()
+	}
+}
+
+// Отказ прохода не убивает цикл: воркер, умерший от временного отказа базы,
+// остановил бы свёртку до конца жизни процесса, пока приём продолжал бы
+// отвечать 200.
+func TestRunПереживаетОтказПрохода(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	w, st, _, sink := newWorker(t, dir)
+
+	// База закрыта — выборка неразобранных отказывает на каждом проходе.
+	if err := st.Close(); err != nil {
+		t.Fatalf("закрытие базы: %v", err)
+	}
+
+	// Второй отказ доказывает, что цикл пережил первый. Разбудить второй проход
+	// без ожидания по часам может только сигнал: тик идёт раз в минуту.
+	twice := sink.expect("fold pass failed", 2)
+	w.Notify()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		w.Run(ctx)
+	}()
+
+	select {
+	case <-twice:
+	case <-time.After(30 * time.Second):
+		t.Fatalf("цикл не пережил отказ прохода:\n%s", sink.dump())
+	}
+	cancel()
+	<-done
+}

internal/store/delivery.go

@@ -54,6 +54,13 @@ type Delivery struct {
 }
 
 // CreateDelivery записывает факт приёма пакета.
+//
+// Через ту же транзакцию с повторами, что и слияние точек, и это не симметрия
+// ради симметрии. Свёртка держит запись всю доставку целиком — измерено 11
+// секунд на 16 тысячах объектов, — а с фоновым воркером конкуренция за базу
+// стала штатной. Одиночный `Exec` пересиживал бы только `busy_timeout`, после
+// чего приём ответил бы `500` по доставке, тело которой уже на диске: доставка
+// исчезла бы из журнала, а телефон её не перешлёт.
 func (s *Store) CreateDelivery(ctx context.Context, d Delivery) error {
 	const q = `
 		INSERT INTO delivery (id, received_at, automation_name, automation_id,
@@ -66,10 +73,13 @@ func (s *Store) CreateDelivery(ctx context.Context, d Delivery) error {
 		headers = "{}"
 	}
 
-	_, err := s.db.ExecContext(ctx, q,
-		d.ID, FormatTime(d.ReceivedAt), d.AutomationName, d.AutomationID,
-		d.Aggregation, d.Period, d.SessionID, d.Bytes, d.SHA256,
-		d.RawPath, d.ParseStatus, d.Points, headers)
+	err := s.inTx(ctx, func(tx *sql.Tx) error {
+		_, err := tx.ExecContext(ctx, q,
+			d.ID, FormatTime(d.ReceivedAt), d.AutomationName, d.AutomationID,
+			d.Aggregation, d.Period, d.SessionID, d.Bytes, d.SHA256,
+			d.RawPath, d.ParseStatus, d.Points, headers)
+		return err //nolint:wrapcheck // обёртка одна, на выходе
+	})
 	if err != nil {
 		return fmt.Errorf("insert delivery: %w", err)
 	}
@@ -148,6 +158,69 @@ func (s *Store) ListDeliveries(ctx context.Context) ([]Delivery, error) {
 	return out, nil
 }
 
+// PendingDelivery — доставка, ожидающая свёртки. Она же курсор обхода: место в
+// журнале задаётся парой `(received_at, id)`, и вызывающему достаточно передать
+// обратно последнюю полученную строку.
+//
+// Метка приёма отдаётся не для порядка (его держит SQL), а для метки отставания:
+// «доставка ждала свёртки дольше N» считается от неё.
+type PendingDelivery struct {
+	ID         string
+	ReceivedAt time.Time
+}
+
+// PendingDeliveries возвращает неразобранные доставки в порядке журнала,
+// строго после курсора. Нулевой курсор означает «с начала».
+//
+// Курсор нужен не ради страниц, а ради завершимости обхода: доставка, у которой
+// не удалось записать даже исход разбора, остаётся `pending`, и выборка без
+// курсора выдавала бы её бесконечно.
+func (s *Store) PendingDeliveries(ctx context.Context, after PendingDelivery, limit int) ([]PendingDelivery, error) {
+	// Сравнение кортежем, а не через OR: развёрнутая форма даёт SCAN по
+	// индексу вместо SEARCH (проверено EXPLAIN QUERY PLAN). Тот же приём уже
+	// применён в LastDerivedLayer.
+	const q = `
+		SELECT id, received_at FROM delivery
+		WHERE parse_status = ? AND (received_at, id) > (?, ?)
+		ORDER BY received_at, id LIMIT ?`
+
+	rows, err := s.db.QueryContext(ctx, q, ParsePending, FormatTime(after.ReceivedAt), after.ID, limit)
+	if err != nil {
+		return nil, fmt.Errorf("select pending deliveries: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var out []PendingDelivery
+	for rows.Next() {
+		var d PendingDelivery
+		var receivedAt string
+		if err := rows.Scan(&d.ID, &receivedAt); err != nil {
+			return nil, fmt.Errorf("scan pending delivery: %w", err)
+		}
+		d.ReceivedAt, err = ParseTime(receivedAt)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, d)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("select pending deliveries: %w", err)
+	}
+	return out, nil
+}
+
+// CountPendingDeliveries возвращает размер задолженности — сколько доставок
+// ждут свёртки. Нужен ровно одной строке лога при старте: сколько сервис должен
+// разобрать, прежде чем витрина станет полной.
+func (s *Store) CountPendingDeliveries(ctx context.Context) (int64, error) {
+	var n int64
+	if err := s.db.GetContext(ctx, &n,
+		`SELECT count(*) FROM delivery WHERE parse_status = ?`, ParsePending); err != nil {
+		return 0, fmt.Errorf("count pending deliveries: %w", err)
+	}
+	return n, nil
+}
+
 // DeliveryStatus возвращает статус разбора доставки.
 func (s *Store) DeliveryStatus(ctx context.Context, id string) (string, error) {
 	var status string

internal/store/errors.go

@@ -1,8 +1,40 @@
 package store
 
-import "errors"
+import (
+	"context"
+	"errors"
+)
 
 // ErrNotFound — записи нет. Граничную ошибку драйвера (sql.ErrNoRows)
 // транслируем в доменную здесь же, у источника, чтобы выше по коду не торчал
 // database/sql.
 var ErrNotFound = errors.New("запись не найдена")
+
+// ErrBusy — база занята, и повторы транзакции этого не пересидели.
+//
+// Доменная ошибка, а не код драйвера: на неё ветвится свёртка. Отказ по
+// занятости не является свойством доставки — работа просто не сделана, и
+// доставка обязана остаться в очереди. Без этого различения конкуренция за
+// базу выводила бы доставку из очереди навсегда.
+var ErrBusy = errors.New("база занята")
+
+// Transient отвечает, вызван ли отказ ОБСТОЯТЕЛЬСТВАМИ, а не данными.
+//
+// Ровно два случая: работу прекратили снаружи и база оказалась занята дольше,
+// чем длятся повторы транзакции. Оба означают «не сделано», а не «не выходит»,
+// поэтому работа обязана остаться к повторению.
+//
+// Определение живёт здесь, в одном месте, и его читают двое: тот, кто пишет
+// исход разбора доставки, и тот, кто классифицирует этот исход в счётчики. Две
+// копии правила разошлись бы, и доставка одновременно осталась бы в очереди и
+// числилась отказавшей.
+//
+// Дедлайн самой операции сюда НЕ входит: не уложившаяся в бюджет работа не
+// уложится в него и в следующий раз, а бесконечный повтор заведомо
+// безнадёжного — это очередь, которая не движется.
+func Transient(err error) bool {
+	if errors.Is(err, context.DeadlineExceeded) {
+		return false
+	}
+	return errors.Is(err, context.Canceled) || errors.Is(err, ErrBusy)
+}

internal/store/migrations/00006_delivery_pending.sql

@@ -0,0 +1,20 @@
+-- +goose Up
+-- Очередью свёртки служит сама таблица: доставка ждёт разбора в статусе
+-- `pending`, а фоновый воркер выбирает такие строки в порядке журнала. Запрос
+-- идёт чаще, чем раз в минуту, а `delivery` растёт примерно на 300 строк в
+-- сутки — без индекса это скан всей таблицы с сортировкой на каждый проход.
+--
+-- Индекс ЧАСТИЧНЫЙ, и это не украшение: в установившемся режиме неразобранных
+-- доставок ноль или одна, поэтому индекс держит ноль-одну строку. Полный
+-- индекс по `parse_status` хранил бы всю историю (сто тысяч строк в год) ради
+-- выборки из одной. SQLite применяет частичный индекс, когда условие запроса
+-- следует из условия индекса — наш случай.
+--
+-- Порядок колонок = порядок журнала, тот же, в котором проигрывает пересборка.
+-- Второй ключ обязателен: `received_at` хранится с секундной точностью, и
+-- доставки одной секунды без него шли бы в неопределённом порядке.
+CREATE INDEX delivery_pending ON delivery (received_at, id)
+    WHERE parse_status = 'pending';
+
+-- +goose Down
+DROP INDEX delivery_pending;

internal/store/pending_test.go

@@ -0,0 +1,153 @@
+package store_test
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"git.vakhrushev.me/av/healthlog/internal/store"
+)
+
+// Очередь свёртки — сама таблица, и порядок её обхода это порядок журнала:
+// `(received_at, id)`, тот же, в котором проигрывает пересборка.
+func TestPendingDeliveriesИдётВПорядкеЖурнала(t *testing.T) {
+	t.Parallel()
+
+	st := open(t)
+	ctx := context.Background()
+	at := ts(t, "2026-08-01T12:00:00Z")
+
+	// Учёт заполняется в порядке, обратном хронологии, и доставки одной секунды
+	// различаются только идентификатором.
+	seedPending(t, st, "d3", at.Add(time.Second))
+	seedPending(t, st, "d2", at)
+	seedPending(t, st, "d1", at)
+
+	got, err := st.PendingDeliveries(ctx, store.PendingDelivery{}, 10)
+	if err != nil {
+		t.Fatalf("PendingDeliveries: %v", err)
+	}
+	want := []string{"d1", "d2", "d3"}
+	if len(got) != len(want) {
+		t.Fatalf("выбрано %d доставок, ожидалось %d", len(got), len(want))
+	}
+	for i, id := range want {
+		if got[i].ID != id {
+			t.Errorf("на месте %d доставка %q, ожидалась %q", i, got[i].ID, id)
+		}
+	}
+}
+
+// Курсор строго возрастает, и обход им конечен: без этого доставка, у которой
+// не удалось записать даже исход разбора, выбиралась бы бесконечно.
+func TestPendingDeliveriesКурсорСтрогоВозрастает(t *testing.T) {
+	t.Parallel()
+
+	st := open(t)
+	ctx := context.Background()
+	at := ts(t, "2026-08-01T12:00:00Z")
+
+	seedPending(t, st, "d1", at)
+	seedPending(t, st, "d2", at)
+
+	first, err := st.PendingDeliveries(ctx, store.PendingDelivery{}, 1)
+	if err != nil {
+		t.Fatalf("PendingDeliveries: %v", err)
+	}
+	if len(first) != 1 || first[0].ID != "d1" {
+		t.Fatalf("первая порция %+v", first)
+	}
+
+	second, err := st.PendingDeliveries(ctx, first[0], 1)
+	if err != nil {
+		t.Fatalf("PendingDeliveries: %v", err)
+	}
+	if len(second) != 1 || second[0].ID != "d2" {
+		t.Fatalf("вторая порция %+v", second)
+	}
+
+	// Доставка, оставшаяся `pending`, за курсором больше не выбирается — именно
+	// на этом стоит завершимость прохода воркера.
+	third, err := st.PendingDeliveries(ctx, second[0], 1)
+	if err != nil {
+		t.Fatalf("PendingDeliveries: %v", err)
+	}
+	if len(third) != 0 {
+		t.Errorf("за последней доставкой выбрано %d строк", len(third))
+	}
+}
+
+// В очередь попадают только неразобранные: свёрнутая доставка из неё выбывает,
+// иначе воркер сворачивал бы весь журнал на каждом проходе.
+func TestPendingDeliveriesБерётТолькоНеразобранные(t *testing.T) {
+	t.Parallel()
+
+	st := open(t)
+	ctx := context.Background()
+	at := ts(t, "2026-08-01T12:00:00Z")
+
+	seedPending(t, st, "d1", at)
+	seedPending(t, st, "d2", at.Add(time.Second))
+	if err := st.FinishParse(ctx, "d1", store.ParseOutcome{Status: store.ParseDone}); err != nil {
+		t.Fatalf("FinishParse: %v", err)
+	}
+
+	n, err := st.CountPendingDeliveries(ctx)
+	if err != nil {
+		t.Fatalf("CountPendingDeliveries: %v", err)
+	}
+	if n != 1 {
+		t.Errorf("задолженность %d, ожидалась 1", n)
+	}
+
+	got, err := st.PendingDeliveries(ctx, store.PendingDelivery{}, 10)
+	if err != nil {
+		t.Fatalf("PendingDeliveries: %v", err)
+	}
+	if len(got) != 1 || got[0].ID != "d2" {
+		t.Errorf("в очереди %+v, ожидалась только d2", got)
+	}
+}
+
+// Правило «отказ обстоятельств, а не данных» живёт в одном месте: его читают и
+// тот, кто пишет исход разбора, и тот, кто классифицирует этот исход.
+func TestTransientРазличаетОбстоятельстваИДанные(t *testing.T) {
+	t.Parallel()
+
+	cases := map[string]struct {
+		err  error
+		want bool
+	}{
+		"работу прекратили снаружи": {context.Canceled, true},
+		"база занята":               {store.ErrBusy, true},
+		"база занята, обёрнута":     {fmt.Errorf("слияние: %w", store.ErrBusy), true},
+		"не уложились в бюджет":     {context.DeadlineExceeded, false},
+		"записи нет":                {store.ErrNotFound, false},
+		"прочее":                    {errors.New("диск отвалился"), false},
+		"ошибки нет":                {nil, false},
+	}
+
+	for name, c := range cases {
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+
+			if got := store.Transient(c.err); got != c.want {
+				t.Errorf("Transient(%v) = %v, ожидалось %v", c.err, got, c.want)
+			}
+		})
+	}
+}
+
+func seedPending(t *testing.T, st *store.Store, id string, at time.Time) {
+	t.Helper()
+
+	err := st.CreateDelivery(context.Background(), store.Delivery{
+		ID: id, ReceivedAt: at, RawPath: id + ".json.gz",
+		SHA256: "-", ParseStatus: store.ParsePending,
+	})
+	if err != nil {
+		t.Fatalf("запись доставки %q: %v", id, err)
+	}
+}

internal/store/tx.go

@@ -51,7 +51,10 @@ func (s *Store) inTx(ctx context.Context, fn func(*sql.Tx) error) error {
 		lastErr = err
 	}
 
-	return fmt.Errorf("транзакция не прошла за %d попыток: %w", txRetries, lastErr)
+	// Занятость называется доменной ошибкой здесь, у источника: выше по коду
+	// не должно торчать ни `sqlite.Error`, ни его коды, а ветвиться на этот
+	// исход нужно — доставка при нём остаётся в очереди.
+	return fmt.Errorf("%w: транзакция не прошла за %d попыток: %v", ErrBusy, txRetries, lastErr) //nolint:errorlint // раскрываем sentinel, причину — намеренно нет
 }
 
 func runTx(ctx context.Context, db *sql.DB, fn func(*sql.Tx) error) error {