Дозакрыты находки ревью по слиянию сущностей

- Правило покрытия получило второй разряд (условный, как у точек), запрет
  вырождения формы и счёт содержательных элементов ряда: скелет из скаляров и
  ряд из null больше не затирают маршрут. Победитель внутри доставки стал
  функцией множества версий — общим помощником с точками, — а провенанс
  поднимается и при совпавшем хеше, иначе отложенная доставка возвращала витрину
  к прежнему содержимому.
- Одно поле не того типа больше не уносит сущность, а пропуски видны в учётной
  записи доставки (миграция 00008, NULL = «не измерялось»); каноническая форма
  считается один раз и вне транзакции; откат бинаря поверх новой схемы отказывает
  на старте; текст ошибки разбора не несёт значений из тела.
- Ревью кода профилем deep (девять проходов) нашло две регрессии и обе закрыты:
  безусловный второй разряд запирал законный досчёт навсегда, а выбор победителя
  был квадратичен по числу присланных версий одного ключа.
This commit is contained in:
av
2026-08-02 16:38:18 +03:00
parent 51a5272c96
commit 8331328134
52 changed files with 4921 additions and 481 deletions
+259 -93
View File
@@ -106,21 +106,23 @@ func clipRefs(refs []EntityRef) []EntityRef {
// entityVersion — версия сущности вместе с тем, что нужно знать при выборе
// победителя.
//
// Разбор и канонизация ОТЛОЖЕНЫ: они нужны только когда хеш разошёлся с
// сохранённым, то есть на одной доставке из сорока четырёх. Считать их сразу
// значило бы разворачивать маршрут (95% веса тренировки, до мегабайта) в дерево
// значений на каждой копии — ровно та форма, от которой разбор тела отказался
// замером (197 МиБ кучи против 54 МиБ на теле 42 МиБ). Хеш при этом считается
// сразу и один раз на доставку: он и есть быстрый путь.
// Всё считается СРАЗУ и один раз на версию, до входа в транзакцию. Ленивость
// здесь была мнимой: хеш всё равно требует полной канонической формы, то есть
// самая дорогая работа платилась на каждой копии и так, а отложенный разбор
// считал ту же форму ВТОРОЙ раз — и делал это внутри транзакции, которая
// открыта `immediate` и повторяется до пяти раз при занятости базы.
//
// Баланс назван честно: на пути разошедшегося хеша (одна доставка из сорока
// четырёх) стало на одну полную канонизацию меньше; на пути совпавшего хеша
// добавился мелкий разбор в map[string]json.RawMessage — проход по телу без
// разворачивания значений. Внутри транзакции для приехавших версий не остаётся
// ничего.
type entityVersion struct {
raw json.RawMessage
hash string
from DeliveryRef
// key и fields заполняются лениво, методом analyze().
raw json.RawMessage
hash string
from DeliveryRef
key []byte
fields canon.Fields
parsed bool
// head — заголовок, который пишется колонками. У сохранённой версии он не
// нужен: она либо побеждает и остаётся как есть, либо замещается целиком.
@@ -128,21 +130,56 @@ type entityVersion struct {
}
func newEntityVersion(e IncomingEntity, from DeliveryRef) (entityVersion, error) {
h, err := canon.Hash(e.Raw)
v, err := analyzeVersion(e.Raw, from)
if err != nil {
return entityVersion{}, fmt.Errorf("хеш сущности: %w", err)
return entityVersion{}, err
}
return entityVersion{raw: e.Raw, hash: h, from: from, head: e}, nil
v.head = e
return v, nil
}
// analyze разбирает версию, если этого ещё не делали.
func (v *entityVersion) analyze() {
if v.parsed {
return
// newStoredVersion собирает версию, прочитанную из витрины.
//
// Каноническая форма здесь НЕ считается, и это существенно: разбор сохранённой
// версии — единственная работа, которая осталась внутри транзакции, открытой
// `immediate`. Замер на тренировке в 168 КБ: полная канонизация с хешем — 4.5 мс
// и 2.3 МБ на 38 тысячах аллокаций, множества ключей — 1.3 мс и 174 КБ на
// тридцати. Хеш сохранённой уже лежит колонкой, а форма нужна ровно в одной
// ветке тай-брейка (равные позиции журнала — та же доставка, свёрнутая
// повторно) и считается там лениво.
func newStoredVersion(raw json.RawMessage, hash string, from DeliveryRef) entityVersion {
return entityVersion{
raw: raw,
hash: hash,
from: from,
fields: canon.Analyze(raw),
}
v.key = canon.SortKey(v.raw)
v.fields = canon.Analyze(v.raw)
v.parsed = true
}
func analyzeVersion(raw json.RawMessage, from DeliveryRef) (entityVersion, error) {
form, hash, err := canon.FormAndHash(raw)
if err != nil {
return entityVersion{}, fmt.Errorf("канонизация сущности: %w", err)
}
return entityVersion{
raw: raw,
hash: hash,
from: from,
key: form,
fields: canon.Analyze(raw),
}, nil
}
// sortKey отдаёт каноническую форму версии, считая её при необходимости.
//
// Ленивость здесь одна на весь файл и нужна ровно сохранённой версии: у неё
// форма требуется только в тай-брейке равных позиций журнала, а стоит она
// втрое дороже разбора и платится под блокировкой записи.
func (v *entityVersion) sortKey() []byte {
if v.key == nil {
v.key = canon.SortKey(v.raw)
}
return v.key
}
// pickEntity выбирает между сохранённой и приехавшей версией.
@@ -177,35 +214,23 @@ func pickEntity(stored, incoming *entityVersion) (takeIncoming, lost bool) {
// Объединение полей отвергнуто там же и по той же причине, что для
// точек, — на живом потоке событие не наступало ни разу, — а из двух
// версий остаётся сохранённая: правило называется «не теряет
// содержания», и приехавшая его теряет. Исход при этом остаётся
// функцией журнала: доставки проигрываются в его порядке.
// содержания», и приехавшая его теряет.
//
// ЗДЕСЬ И ТОЛЬКО ЗДЕСЬ исход зависит от порядка свёртки, а не от
// журнала: в витрине лежит победитель прошлых слияний, а не все
// кандидаты истории, и «сохранённая выигрывает» означает разный итог
// при разном порядке. Порядок свёртки журналу не равен — доставка,
// получившая ErrBusy, остаётся `pending` и сворачивается следующим
// проходом, — так что живой приём и пересборка на несравнимых версиях
// законно расходятся. Это единственная точка, где витрина не является
// функцией множества доставок; она названа вслух в architecture.md, и
// счётчик удержаний ниже — единственное, что о ней сообщает.
return false, true
default:
return laterInJournal(stored, incoming), false
}
}
// pickWithinDelivery выбирает между двумя версиями одного ключа ВНУТРИ одной
// доставки. Второй возврат — различается ли их содержание вообще.
//
// Отдельно от pickEntity, и не ради симметрии: «сохранённой» версии здесь нет,
// есть только порядок элементов в JSON-массиве, а он нестабилен. Правило
// «остаётся первая встреченная» сделало бы исход функцией порядка на проводе,
// поэтому при равном и при несравнимом содержании решает тотальный порядок
// канонических форм.
func pickWithinDelivery(a, b *entityVersion) (takeB, differs bool) {
switch v := compareEntities(a, b); v {
case entityIncomingRicher:
return true, true
case entityStoredRicher:
return false, true
case entityIncomparable:
return laterInJournal(a, b), true
default:
return laterInJournal(a, b), false
}
}
// entityVerdict — как соотносится СОДЕРЖАНИЕ двух версий одной сущности.
// Нумерация с единицы: нулевое значение не должно выглядеть как «равны».
type entityVerdict int
@@ -221,9 +246,6 @@ const (
)
func compareEntities(stored, incoming *entityVersion) entityVerdict {
stored.analyze()
incoming.analyze()
storedCovers := stored.fields.Covers(incoming.fields)
incomingCovers := incoming.fields.Covers(stored.fields)
@@ -250,58 +272,135 @@ func laterInJournal(stored, incoming *entityVersion) bool {
if incoming.from.before(stored.from) {
return false
}
return bytes.Compare(incoming.key, stored.key) < 0
return bytes.Compare(incoming.sortKey(), stored.sortKey()) < 0
}
// dedupeEntities сворачивает версии одного ключа ВНУТРИ доставки тем же
// правилом — до сравнения с сохранённой.
// entityDominates говорит, СТРОГО ли a превосходит b по содержанию: покрывает и
// не покрывается в ответ.
//
// Без этого исход зависел бы от того, как написан цикл: карта по ключу дала бы
// победу последнему элементу массива мимо правила полноты, а порядок элементов
// в JSON-массиве нестабилен.
//
// Счётчик здесь считает СИММЕТРИЧНО — «в одном теле приехали две версии одного
// ключа с разным содержанием», — а не «приехавшая обеднена». Внутри доставки
// «сохранённой» версии не существует, есть только порядок элементов массива, и
// счётчик, зависящий от него, наблюдал бы событие через раз.
func dedupeEntities(versions []entityVersion) ([]entityVersion, int, []EntityRef) {
type slot struct {
v entityVersion
pos int
}
// Строгость обязательна. Covers — предпорядок, а не строгий порядок: две версии
// могут покрывать друг друга взаимно (тот же набор ключей, другие значения), и
// отбрасывание «всего, что кем-то покрыто» опустошило бы множество, потеряв обе.
func entityDominates(a, b entityVersion) bool {
return a.fields.Covers(b.fields) && !b.fields.Covers(a.fields)
}
byKey := make(map[EntityRef]slot, len(versions))
// entityLess — тотальный порядок на версиях равного содержания.
//
// Сперва каноническая форма, потом ИСХОДНЫЕ БАЙТЫ. Второй разряд не украшение:
// у сущностей версии с равной формой не схлопываются (в отличие от точек, где
// это делает дедупликация по ключу), а у HAE порядок ключей в JSON и запись
// числа нестабильны — то есть без него минимум неединствен, и в витрину лёг бы
// тот элемент, что стоял в массиве раньше. Порядок элементов на проводе не
// имеет права решать, какие байты хранятся.
func entityLess(a, b entityVersion) bool {
if c := bytes.Compare(a.key, b.key); c != 0 {
return c < 0
}
return bytes.Compare(a.raw, b.raw) < 0
}
// dedupeEntities сворачивает версии одного ключа ВНУТРИ доставки — до сравнения
// с сохранённой.
//
// Победитель здесь — функция МНОЖЕСТВА версий, а не порядка элементов массива:
// сперва отбрасываются строго покрытые, среди оставшихся берётся минимум
// тотального порядка. Попарная свёртка была неверна ровно так же, как она была
// неверна для точек: покрытие — частичный порядок, тай-брейк — тотальный, и
// вместе они дают нетранзитивную победу, при которой [A,B,C] и [B,C,A] дают
// разных победителей.
//
// Версии с СОВПАВШЕЙ канонической формой схлопываются ДО выбора победителя, и
// это не оптимизация ради красоты: выбор квадратичен по числу кандидатов, а их
// число приходит из чужого тела. Точки схлопываются так же и в том же месте
// (см. mergePoints). Внутри схлопнутой группы остаются минимальные байты —
// тот же второй разряд тотального порядка, что и между группами.
//
// Второй возврат — счётчик «в одном теле приехали версии одного ключа с РАЗНЫМ
// содержанием», симметричный по построению: считаются кандидаты, чья форма
// отличается от формы победителя. По форме, а не по байтам: порядок ключей у
// HAE нестабилен и дребезг последнего разряда тоже, так что побайтовый счётчик
// срабатывал бы на норме потока и стал бы неотличим от шума ровно тогда, когда
// понадобился бы.
//
// Счётчик отдельный от «удержаний», а не общий с ними. Две версии в одном теле —
// это НЕ потеря содержания: победитель ложится в витрину целиком, и удерживать
// нечего. Смешивать их значило бы отвечать одним числом на два вопроса, которые
// лечатся по-разному, — а на число удержаний опирается единственный контроль
// того, что правило покрытия не стало слишком строгим.
func dedupeEntities(ctx context.Context, versions []entityVersion) ([]entityVersion, int, []EntityRef, error) {
byKey := make(map[EntityRef][]entityVersion, len(versions))
order := make([]EntityRef, 0, len(versions))
held := 0
var heldAt []EntityRef
for _, v := range versions {
ref := EntityRef{Kind: v.head.Kind, ID: v.head.ID}
prev, seen := byKey[ref]
if !seen {
byKey[ref] = slot{v: v, pos: len(order)}
if _, seen := byKey[ref]; !seen {
order = append(order, ref)
continue
}
takeB, differs := pickWithinDelivery(&prev.v, &v)
if differs {
held++
if len(heldAt) < maxEntityRefsReported {
heldAt = append(heldAt, ref)
}
}
winner := prev.v
if takeB {
winner = v
}
byKey[ref] = slot{v: winner, pos: prev.pos}
byKey[ref] = append(byKey[ref], v)
}
out := make([]entityVersion, 0, len(order))
diverging := 0
var divergingAt []EntityRef
for _, ref := range order {
out = append(out, byKey[ref].v)
cands, dropped := collapseEqualForms(byKey[ref])
winner, _, err := pickBest(ctx, cands, entityDominates, entityLess)
if err != nil {
return nil, 0, nil, err
}
out = append(out, cands[winner])
// Схлопнутые копии победителя различием не считаются: их форма ему
// равна. Считаются все прочие — и оставшиеся кандидаты, и те, что
// схлопнулись в них.
differing := 0
for i := range cands {
if i != winner {
differing += 1 + dropped[i]
}
}
if differing > 0 {
diverging += differing
if len(divergingAt) < maxEntityRefsReported {
divergingAt = append(divergingAt, ref)
}
}
}
return out, held, heldAt
return out, diverging, divergingAt, nil
}
// collapseEqualForms схлопывает версии с одинаковой канонической формой в одну,
// оставляя минимальные исходные байты. Второй возврат — сколько копий сложилось
// в каждого оставшегося кандидата (нужно счётчику различий).
//
// Схлопывание обязательно, а не желательно: без него тело с двадцатью тысячами
// повторов одного `id` даёт четыреста миллионов сравнений покрытия, каждое с
// обходом массивов. Тело в пределах приёма такое вмещает.
func collapseEqualForms(versions []entityVersion) ([]entityVersion, []int) {
byForm := make(map[string]int, len(versions))
out := make([]entityVersion, 0, len(versions))
dropped := make([]int, 0, len(versions))
for _, v := range versions {
form := string(v.key)
i, seen := byForm[form]
if !seen {
byForm[form] = len(out)
out = append(out, v)
dropped = append(dropped, 0)
continue
}
dropped[i]++
if bytes.Compare(v.raw, out[i].raw) < 0 {
// Байты решают внутри группы ровно так же, как между группами:
// порядок элементов на проводе не имеет права выбирать содержимое.
// Заголовок едет вместе с байтами — он от них производен.
out[i] = v
}
}
return out, dropped
}
// mergeEntities сливает сущности одной секции с сохранёнными.
@@ -320,11 +419,31 @@ func mergeEntities(ctx context.Context, tx *sql.Tx, table string, versions []ent
continue
}
// Хеш — детектор изменений: совпал, значит писать нечего, и содержимое
// сохранённой сущности читать не приходится вовсе. Тренировка
// переприсылается каждой доставкой, пока не доедет маршрут, — на живом
// архиве 44 копии дают три различных содержимых.
// Хеш — детектор изменений: совпал, значит содержимое то же, и читать
// его не приходится вовсе. Тренировка переприсылается каждой доставкой,
// пока не доедет маршрут, — на живом архиве 44 копии дают три различных
// содержимых.
//
// Но провенанс при этом обновить НАДО. Сохранённая позиция журнала
// участвует в тай-брейке «содержание равно», и если в ней осталась
// первая свёрнутая копия вместо победителя журнала, отложенная доставка
// вернёт витрину к прежнему содержимому — то есть живая витрина
// разойдётся с пересборкой, молча и в содержимом тренировки.
//
// Предел назван вслух: обновляется провенанс, но НЕ байты. При
// совпавшей канонической форме в витрине остаются байты той доставки,
// что свернулась первой, — а порядок ключей у HAE нестабилен, значит у
// живого приёма и пересборки они могут различаться. Отпечаток этого не
// различает (он считает по канонической форме), содержания не теряется
// ничего, а переписывать мегабайтный маршрут на каждой из двадцати
// шести присылок ради выбора между эквивалентными литералами — цена
// несоразмерная.
if stored.hash == v.hash {
if stored.from.before(v.from) {
if err := touchEntityProvenance(ctx, tx, table, v); err != nil {
return 0, 0, nil, err
}
}
continue
}
@@ -332,7 +451,7 @@ func mergeEntities(ctx context.Context, tx *sql.Tx, table string, versions []ent
if err != nil {
return 0, 0, nil, err
}
prev := entityVersion{raw: storedRaw, hash: stored.hash, from: stored.from}
prev := newStoredVersion(storedRaw, stored.hash, stored.from)
takeIncoming, lost := pickEntity(&prev, &v)
if lost {
@@ -352,6 +471,44 @@ func mergeEntities(ctx context.Context, tx *sql.Tx, table string, versions []ent
return written, held, heldAt, nil
}
// touchEntityProvenance поднимает провенанс сущности до более поздней доставки
// журнала, не трогая содержимое.
//
// `updated_at` НЕ двигается, и это отдельное решение, а не экономия. Тренировка
// приезжает до двадцати шести раз; бамп метки на каждой сделал бы её меткой
// касания строки, а не изменения содержимого, и потребитель запроса «что
// изменилось с момента X» получил бы двадцать шесть ложных изменений,
// неотличимых от настоящего досчёта. Провенанс несёт собственную метку —
// времени приёма своей доставки, — и для тай-брейка её достаточно.
//
// Счётчик записанных сущностей такое обновление тоже не увеличивает: он считает
// СОДЕРЖИМОЕ витрины, и сравнимость его с прежними замерами важнее учёта
// обновлённой ссылки.
func touchEntityProvenance(ctx context.Context, tx *sql.Tx, table string, v entityVersion) error {
q := `UPDATE ` + table + ` SET delivery_id = ?, delivery_received_at = ?` + entityWhere(table)
args := append([]any{v.from.ID, FormatTime(v.from.ReceivedAt)},
entityKeyArgs(table, v.head.Kind, v.head.ID)...)
res, err := tx.ExecContext(ctx, q, args...)
if err != nil {
return fmt.Errorf("update %s provenance: %w", table, err)
}
// Строка гарантированно существует: её заголовок прочитан этой же
// транзакцией десятью строками выше. Ноль означал бы, что ключ собран не
// теми колонками, — а провенанс в отпечаток витрины не входит, значит
// молчаливый промах не поймает ни один оракул сходимости. Соседи по файлу
// (FinishParse, MarkSealed) проверяют по той же причине.
n, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("update %s provenance: %w", table, err)
}
if n == 0 {
return fmt.Errorf("update %s provenance: %w", table, ErrNotFound)
}
return nil
}
type storedEntityHead struct {
hash string
from DeliveryRef
@@ -406,11 +563,20 @@ func entityWhere(table string) string {
return ` WHERE id = ?`
}
func queryEntity(ctx context.Context, tx *sql.Tx, q, table, kind, id string) *sql.Row {
// entityKeyArgs — аргументы к entityWhere. Живут рядом с ним намеренно: число
// `?` в тексте и длина этого среза обязаны меняться вместе, а компилятор их
// соответствия не видит. Промах даст ошибку SQLite внутри транзакции слияния,
// то есть на пути, который повторяется до пяти раз и оканчивается `failed` у
// доставки, а не отказом сборки.
func entityKeyArgs(table, kind, id string) []any {
if table == recordTable {
return tx.QueryRowContext(ctx, q, kind, id)
return []any{kind, id}
}
return tx.QueryRowContext(ctx, q, id)
return []any{id}
}
func queryEntity(ctx context.Context, tx *sql.Tx, q, table, kind, id string) *sql.Row {
return tx.QueryRowContext(ctx, q, entityKeyArgs(table, kind, id)...)
}
func writeEntity(ctx context.Context, tx *sql.Tx, table string, v entityVersion, now time.Time) error {