Идемпотентный импорт и активация ревизии

Таблицы готовы хранить несколько ревизий документа. Теперь добавим импорт: прочитаем фрагменты, получим векторы и переключим документ на завершённое состояние. Главная задача этого урока — согласованность. Читатель не должен увидеть заголовок новой страницы рядом с половиной старого текста только потому, что загрузка очередного вектора ещё не закончилась.

В lesson-05 появляется ingest.py. Он использует функции предыдущих этапов и не меняет формат корпуса. После выполнения читателем ожидается активация трёх документов; повторный импорт того же состояния должен сохранить текущую ревизию. База, модель и программа автором не запускались, поэтому ниже объясняется ожидаемый контракт операции.

Подготовка вне транзакции

Сначала split_doc() строит фрагменты, а revision() вычисляет отпечаток состояния. Затем адаптер получает список заголовков и текстов. Это потенциально медленная часть: локальная модель может загружаться, обрабатывать данные или возвращать ошибку. Пока она работает, мы ещё не удерживаем блокировку строки документа в PostgreSQL.

Такой порядок сокращает время транзакции, но требует хранить подготовленное состояние в памяти до вставки. Для маленького корпуса это приемлемо. Большой импорт нуждается в пакетах, промежуточном хранилище и ограничении объёма одновременно подготовленных векторов. Перенос сетевого ожидания внутрь транзакции не решает проблему памяти и добавляет новые задержки для других операций.

Если адаптер отказал, активный указатель документа остаётся прежним. Ошибка получения векторов не должна превращаться в публикацию пустой ревизии. Именно поэтому подготовка заканчивается до изменения видимого состояния, а размер и количество векторов проверяются самим адаптером.

Запись одной ревизии

Рассмотрим полный файл импорта. Контекст соединения psycopg завершает транзакцию при успешном выходе и откатывает изменения при исключении. Функция возвращает результат операции, но не выдаёт отдельно выполненные SQL-команды за уже сохранённый документ до завершения контекста.

from chunks import load_corpus, split_doc, revision
from embeddings import get_embedder,vector_literal
from db import connect
def ingest(doc,embedder):
    parts=split_doc(doc);rev=revision(doc,embedder.space)
    texts=[p['heading']+'\n'+p['text'] for p in parts]
    vectors=embedder.embed(texts) # model work completes before the transaction
    with connect() as conn:
        conn.execute('INSERT INTO documents(doc_id,url,title,visibility) VALUES(%s,%s,%s,%s) ON CONFLICT(doc_id) DO NOTHING',
                     (doc['id'],doc['url'],doc['title'],doc['visibility']))
        old=conn.execute('SELECT * FROM documents WHERE doc_id=%s FOR UPDATE',(doc['id'],)).fetchone()
        if old['active_revision']==rev and old['active_space']==embedder.space and not old['deleted']:
            return {'status':'unchanged','revision':rev}
        conn.execute('INSERT INTO revisions(doc_id,revision,space,fragment_count) VALUES(%s,%s,%s,%s) ON CONFLICT DO NOTHING',
                     (doc['id'],rev,embedder.space,len(parts)))
        for p,vec in zip(parts,vectors):
            conn.execute('INSERT INTO chunks(doc_id,revision,space,part,heading,body,embedding) VALUES(%s,%s,%s,%s,%s,%s,%s::vector) ON CONFLICT DO NOTHING',
                         (doc['id'],rev,embedder.space,p['part'],p['heading'],p['text'],vector_literal(vec)))
        count=conn.execute('SELECT count(*) AS n FROM chunks WHERE doc_id=%s AND revision=%s AND space=%s',
                           (doc['id'],rev,embedder.space)).fetchone()['n']
        if count!=len(parts):raise ValueError('incomplete prepared revision')
        conn.execute('UPDATE documents SET url=%s,title=%s,visibility=%s,active_revision=%s,active_space=%s,deleted=false WHERE doc_id=%s',
                     (doc['url'],doc['title'],doc['visibility'],rev,embedder.space,doc['id']))
        return {'status':'activated','revision':rev,'fragments':count}
if __name__=='__main__':
    embedder=get_embedder()
    for doc in load_corpus():print(doc['id'],ingest(doc,embedder))

Первая вставка создаёт строку documents, если идентификатор ещё не существует. Затем SELECT ... FOR UPDATE блокирует эту строку на время операции. Два импортёра одного документа не должны одновременно переключать её метаданные и активную ревизию. Для разных документов блокировки остаются независимыми.

Блокировка не определяет, какой исходный файл «новее». Если два процесса подготовили разные тексты заранее, позже закончившийся процесс может активировать более старый материал. В этой лаборатории предполагается один оператор импорта и отсутствие параллельных обновлений корпуса. Для промышленного редакционного процесса потребуются очередь, номер редакционной версии или проверка ожидаемой текущей ревизии.

Идемпотентность и повторение

Если активная ревизия, пространство и признак удаления уже совпадают, функция возвращает unchanged. Повторное выполнение не создаёт второй комплект одинаковых строк. Составные первичные ключи и ON CONFLICT DO NOTHING дополнительно защищают записи ревизии и фрагментов от простого дублирования.

Но получение векторов здесь происходит раньше проверки активного состояния. Повторный импорт может снова потратить время модели. Это цена простой последовательности: она идемпотентна по результату хранения, а не бесплатна по вычислениям. Дальнейшая оптимизация могла бы сначала читать отпечаток и аккуратно обрабатывать гонку между предварительной проверкой и записью.

Не следует автоматически превращать любой конфликт в успех. Например, два документа с разными идентификаторами не могут иметь один адрес из-за уникального ограничения. Такой конфликт требует исправления корпуса. Молчаливое присвоение чужого адреса разрушило бы связь источника и документа.

Проверка полноты перед переключением

После вставки всех фрагментов функция считает строки выбранной ревизии. Количество должно совпасть с подготовленным списком. Это простая проверка целостности этапа: активировать документ с недостающим фрагментом нельзя. Если условие нарушено, исключение откатывает текущую транзакцию.

Счётчик не доказывает буквальное совпадение каждого ранее существовавшего фрагмента. Мы предполагаем, что отпечаток состояния получен тем же алгоритмом и строки не были вручную изменены. Для среды с посторонними писателями нужны дополнительные права, контроль содержимого или восстановление ревизии целиком. Один хеш в названии не защищает таблицу от произвольного UPDATE.

Последний UPDATE переключает метаданные и оба активных указателя одной SQL-операцией. До фиксации транзакции читатели продолжают видеть прежнее состояние. После фиксации новый запрос может увидеть новый заголовок, публичность, ревизию и пространство вместе. Это свойство согласованной записи, а не обещание, что уже начатые чтения автоматически обновятся.

Возврат к удалённому документу

Поле deleted участвует в проверке повторного состояния. Если документ был скрыт, совпадения отпечатка недостаточно для unchanged: явный импорт снова активирует его и снимает признак удаления. Такое поведение следует считать редакционным действием, а не случайной побочной особенностью алгоритма.

Для другого процесса можно выбрать противоположную политику: удалённый документ восстанавливается только отдельной командой. Тогда потребуется изменить условие и договор оператора. Важно, чтобы обычное переиндексирование не неожиданно возвращало материалы, которые редактор намеренно убрал из публичного корпуса.

Наш импорт не удаляет документы, отсутствующие в JSON-файле. Это тоже явная граница. Исчезновение строки из входного списка ещё не сообщает, произошло ли редакционное удаление, временный сбой чтения или неполная выгрузка. В десятом уроке введём отдельную операцию скрытия вместо опасного предположения о полном зеркале.

Рассмотрим отказ в середине записи. Новые строки могли быть подготовлены в транзакции, но проверка количества обнаружила неполный набор. Исключение приводит к откату изменений этого соединения, поэтому действующие указатели не должны переключиться на частично записанную ревизию. Это ожидаемое свойство устройства функции, а не результат выполненного опыта. Уже существовавшие записи базы откат не уничтожает. Если оператор раньше записал ошибочную ревизию обходным способом, один только повтор импорта не исправляет произвольные нарушения истории. Наш договор предполагает, что таблицами управляют согласованные функции. Прямая ручная правка требует отдельного разбора и восстановления инвариантов, особенно если число строк совпало, а их содержание отличается.

После самостоятельного импорта сравните возвращаемые статусы первого и повторного прохода. Ожидается одинаковая активная ревизия при неизменном состоянии. Не публикуем выдуманные отпечатки или время работы: их нужно получить в вашей среде. Следующий урок прочитает эти записи по расстоянию и покажет, как ограничить источники до формирования выдачи.

Исходники и документация

Оглавление серии · Все состояния проекта. Для этого урока откройте папку lesson-05.

pgvector, транзакции Psycopg, Ollama Embed, структурированные ответы.

Примеры подготовлены как исходники. Их выполнение, замеры, тесты и публикация при написании не проводились. Описанные результаты ожидаемые.