Compare commits
53 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f092723a1a | |||
| 8b6f2e2c0c | |||
| 55bb0e7cc0 | |||
| f494fca375 | |||
| 6cde3c4cc9 | |||
| 5447aa45d1 | |||
| 2da91c8bc1 | |||
| cbaf5a4918 | |||
| 29e66d01fb | |||
| 564d0ac688 | |||
| 46c73814f6 | |||
| 0f2117d8a1 | |||
| e7ab44041c | |||
| 45dd4ccdb8 | |||
| 75506b9fb7 | |||
| 2de05922c5 | |||
| e6d064b6cf | |||
| d1a57ee199 | |||
| 08dbde5731 | |||
| a75f29a231 | |||
| c858b1eec3 | |||
| 2cc2249625 | |||
| 0a6cfe8f0e | |||
| 4593c30076 | |||
| 9150dbab1f | |||
| 8d26429021 | |||
| ab9e8375cb | |||
| 234a8c23c5 | |||
| a037cc4bb8 | |||
| c235c959e6 | |||
| 5201fc4055 | |||
| 77d12a6f8c | |||
| 592becc486 | |||
| 96b20b7d58 | |||
| 3ca246432d | |||
| 29a1f8dc27 | |||
| d49729bf1a | |||
| 12dbd16f7b | |||
| 8fa28ada45 | |||
| 881bdbc5dc | |||
| f63ae0a523 | |||
| 50d3b72d0b | |||
| aa306c4159 | |||
| 9214729d2f | |||
| 5c7100a67c | |||
| a002d6038b | |||
| 2fd82f2939 | |||
| 48fe3447ad | |||
| f170024e75 | |||
| 38a229a107 | |||
| 35372dc06a | |||
| 525f9d6d82 | |||
| 7bebd91ab6 |
59
ISSUE_TEMPLATE/bug_report.yaml
Normal file
59
ISSUE_TEMPLATE/bug_report.yaml
Normal file
@@ -0,0 +1,59 @@
|
||||
name: "🐛 Сообщить об ошибке"
|
||||
description: "Создайте отчёт, чтобы помочь нам исправить проблему"
|
||||
title: "[BUG]: "
|
||||
labels: ["bug"]
|
||||
body:
|
||||
- type: markdown
|
||||
attributes:
|
||||
value: |
|
||||
## Спасибо, что помогаете сделать Futriix лучше!
|
||||
Пожалуйста, опишите проблему максимально подробно.
|
||||
|
||||
- type: textarea
|
||||
id: description
|
||||
attributes:
|
||||
label: "Описание ошибки"
|
||||
description: "Чётко и ясно опишите, в чём заключается проблема."
|
||||
placeholder: "Например: При попытке сохранить документ с полем типа 'timestamp' возникает паника..."
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: steps
|
||||
attributes:
|
||||
label: "Шаги для воспроизведения"
|
||||
description: "Опишите пошагово, что нужно сделать, чтобы увидеть ошибку."
|
||||
placeholder: |
|
||||
1. Запустить СУБД с конфигом...
|
||||
2. Отправить запрос...
|
||||
3. Увидеть ошибку...
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: input
|
||||
id: version
|
||||
attributes:
|
||||
label: "Версия Futriix"
|
||||
description: "Какую версию вы используете? (Например, 2.0 i²)"
|
||||
placeholder: "v2.0"
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: dropdown
|
||||
id: os
|
||||
attributes:
|
||||
label: "Операционная система"
|
||||
options:
|
||||
- Linux
|
||||
- Windows
|
||||
- macOS
|
||||
- Другая
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: logs
|
||||
attributes:
|
||||
label: "Логи или вывод терминала"
|
||||
description: "Вставьте сюда любые сообщения об ошибках или логи (можно в виде текста)."
|
||||
render: shell
|
||||
36
ISSUE_TEMPLATE/feature_request.yaml
Normal file
36
ISSUE_TEMPLATE/feature_request.yaml
Normal file
@@ -0,0 +1,36 @@
|
||||
name: "🚀 Предложить новую функцию"
|
||||
description: "Поделитесь идеей, которая улучшит Futriix"
|
||||
title: "[FEATURE]: "
|
||||
labels: ["enhancement"]
|
||||
body:
|
||||
- type: markdown
|
||||
attributes:
|
||||
value: |
|
||||
## Мы открыты к вашим идеям!
|
||||
Расскажите, какую возможность вы хотели бы видеть.
|
||||
|
||||
- type: textarea
|
||||
id: idea
|
||||
attributes:
|
||||
label: "Ваша идея"
|
||||
description: "Опишите, что бы вы хотели добавить или изменить."
|
||||
placeholder: "Было бы здорово реализовать поддержку индексов для поля 'name'..."
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: benefit
|
||||
attributes:
|
||||
label: "Какую проблему это решает?"
|
||||
description: "Объясните, зачем это нужно вам и другим пользователям."
|
||||
placeholder: "Это ускорит поиск по документам в 3 раза..."
|
||||
|
||||
- type: textarea
|
||||
id: lua_example
|
||||
attributes:
|
||||
label: "Пример использования (если связано с Lua-плагином)"
|
||||
description: "Если ваша идея касается системы плагинов, приведите пример кода на Lua."
|
||||
render: lua
|
||||
placeholder: |
|
||||
-- Например, как бы вы хотели использовать новый движок?
|
||||
local my_engine = futriix.new_engine("custom")
|
||||
27
ISSUE_TEMPLATE/question.yaml
Normal file
27
ISSUE_TEMPLATE/question.yaml
Normal file
@@ -0,0 +1,27 @@
|
||||
name: "❓ Задать вопрос"
|
||||
description: "Нужна помощь или совет по использованию Futriix"
|
||||
title: "[QUESTION]: "
|
||||
labels: ["question"]
|
||||
body:
|
||||
- type: markdown
|
||||
attributes:
|
||||
value: |
|
||||
## Мы здесь, чтобы помочь!
|
||||
Не стесняйтесь задавать любые вопросы по установке, настройке или использованию СУБД.
|
||||
|
||||
- type: textarea
|
||||
id: question
|
||||
attributes:
|
||||
label: "Ваш вопрос"
|
||||
description: "Чётко сформулируйте, что именно вас интересует."
|
||||
placeholder: "Как правильно настроить кластер из двух узлов?"
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: context
|
||||
attributes:
|
||||
label: "Дополнительный контекст"
|
||||
description: "Расскажите, что вы уже попробовали, или приложите фрагмент кода/конфига."
|
||||
placeholder: "Мой конфиг выглядит так: ..."
|
||||
render: yaml
|
||||
110
README.md
110
README.md
@@ -54,7 +54,7 @@
|
||||
<li><a href="#триггеры">Триггеры</a></li>
|
||||
<li><a href="#сжатие-данных">Сжатие данных</a></li>
|
||||
<li><a href="#графический-интерфейс">Графический интерфейс</a></li>
|
||||
<li><a href="#дорожная-карта">Дорожная карта</a></li>
|
||||
<li><a href="#план-развития">План развития</a></li>
|
||||
<li><a href="#контакты">Контакты</a></li>
|
||||
</ol>
|
||||
<!-- </details> -->
|
||||
@@ -63,12 +63,13 @@
|
||||
## О проекте
|
||||
|
||||
> [!CAUTION]
|
||||
> **ALPHA VERSION**<br><br>**Проект стабилен в тестовых сценариях, но категорически не рекомендуется для промышленного использования до выхода версии 3.0. Мы открыты к обратной связи и тестированию!**
|
||||
> **ALPHA VERSION**<br><br>**Проект достаточно стабилен в тестовых сценариях, но категорически не рекомендуется для промышленного использования до выхода версии 3.0. Мы открыты для предложений, и рады обратной связи!**
|
||||
|
||||
futriix - это легковесная, распределённая NOSQL, использующая алгоритмы неблокирующей синхронизации - `wait-free` и `lock-free` in-memory СУБД, реализованная на языке Go с поддержкой плагинов на языке lua использующая алгоритм консенсуса Raft. <br>
|
||||
По своей сути, Futriis является лёгкой HTAP-СУБД с уклоном в OLTP благодаря wait-free структурам и ACID-транзакциям, а также мощными OLAP-возможностями через индексы, триггеры, Lua-плагины и аналитические функции. Это позволяет обрабатывать и анализировать данные в режиме, близком к реальному времени, в первую очередь для эксплуатации в замкнутых программных средах под управлением ОС на базе ядра Illumos и Solaris (OpenIndiana Hipster, Oracle Solaris). Архитектурное решение использовать отдельные индексы от данных, атомарные операции и версионирование позволяет эффективно обрабатывать смешанную нагрузку без необходимости применять отдельные системы для OLTP и OLAP.<br>
|
||||
По своей сути, Futriis является лёгкой HTAP-СУБД с уклоном в OLTP благодаря wait-free структурам и ACID-транзакциям, а также мощными OLAP-возможностями через индексы, триггеры, Lua-плагины и аналитические функции. Это позволяет обрабатывать и анализировать данные в режиме, близком к реальному времени, в первую очередь для эксплуатации в замкнутых программных средах под управлением ОС на базе ядра Illumos и Solaris (OpenIndiana Hipster, Oracle Solaris). <br>
|
||||
Архитектурное решение использовать отдельные индексы от данных, атомарные операции и версионирование позволяет эффективно обрабатывать смешанную нагрузку без необходимости применять отдельные системы для OLTP и OLAP.<br>
|
||||
В futriix реализован WUI-интерфейс для удобства её администрирования и эксплуатации.
|
||||
Но она также совместима с популярными Linux-дистрибутивами (Debian,Ubuntu,Fedora), т.е. операционными системами построенными на базе ядра Linux.
|
||||
Кроме того она совместима с популярными Linux-дистрибутивами (Debian,Ubuntu,Fedora).
|
||||
|
||||
<p align="right">(<a href="#readme-top">К началу</a>)</p>
|
||||
|
||||
@@ -85,14 +86,16 @@ futriix - это легковесная, распределённая NOSQL, и
|
||||
|
||||
* **База Данных(БД)** - это структурированное, организованное хранилище данных, которое позволяет удобно собирать, хранить, управлять и извлекать информацию.
|
||||
* **Система Управления Базами Данных(СУБД)** - это программное обеспечение, которое позволяет создавать, управлять и взаимодействовать с базами данных
|
||||
* **Таппл (Tapple)** - аналог базы данных в РСУБД
|
||||
* **Слайс (Slice)** - аналог таблицы
|
||||
* **Кортеж (Tuple)** - аналог записи в таблице
|
||||
* **Слайс (Slice-в пер.с англ.слой)**- в терминах субд Futriix имеет два значения:
|
||||
* **1.** Cиноним "база данных"
|
||||
* **2.** Термин пришедший из терминологии суд MongoDB, а именно: Логический и физически изолированный фрагмент коллекции документов, полученный в результате горизонтального партиционирования (шардирования) и размещенный на определенном узле кластера с целью масштабирования производительности и объема данных.
|
||||
* **Коллекция (Collection)** - аналог таблицы
|
||||
* **Поле (Field)** - аттрибут документов
|
||||
* **Кортеж (Tuple)** - вложенный документ в субд Futriix
|
||||
* **Мультимодельная СУБД** - это СУБД, которая объединяет в себе поддержку нескольких моделей данных (реляционной, документной, графовой, ключ-значение и др.) в рамках единого интегрированного ядра.
|
||||
* **Резидентная СУБД** - это СУБД, которая работает непрерывно в оперативной памяти (RAM).
|
||||
* **Инстанс** - это запущенный экземляр базы данных.
|
||||
* **Узел (хост,нода,шард)** - это отдельный сервер (физический или виртуальный), который является частью кластера или распределенной системы и выполняет часть общей работы.
|
||||
* **Слайс (от англ. "slice"-слой)** - это логический и физически изолированный фрагмент коллекции документов, полученный в результате горизонтального партиционирования (шардирования) и размещенный на определенном узле кластера с целью масштабирования производительности и объема данных.
|
||||
* **Репликасет** - это группа серверов СУБД, объединенных в отказоустойчивую конфигурацию, где один узел выполняет роль первичного (принимающего операции записи), а один или несколько других - роль вторичных (синхронизирующих свои данные с первичным и обслуживающих чтение), с автоматическим переизбранием первичного узла в случае его сбоя.
|
||||
* **Временная метка (time stamp)** - это автоматически фиксируемое значение времени создания, изменения или удаления объекта в СУБД, представленное в формате Unix-миллисекунд (количество миллисекунд, прошедших с 1 января 1970 года) с возможностью отображения в человекочитаемом виде YYYY-MM-DD HH:MM:SS.mmm. Необходима для понимание последовательности операций при сбоях, выявления аномалий в поведении системы,архивация старых записей, отслеживание несанкционированных изменений, автоматический выход из сессий с ограниченным временем действия, Статистика активности за период)
|
||||
* **OLTP (Online Transactional Processing-Онлайн обработка транзакций)**- это технология обработки транзакций в режиме реального времени. Её основная задача заключается в обеспечении быстрого и надёжного выполнения операций, которые происходят ежесекундно в бизнесе. Они обеспечивают быстрое выполнение операций вставки, обновления и удаления данных, поддерживая целостность и надежность транзакций.
|
||||
@@ -364,20 +367,27 @@ done
|
||||
> СУБД futriiX полностью совместима по синтаксису запросов с NOSQL-субд MongoDB.
|
||||
> Вы можете использовать знакомые NoSQL-конструкции и драйверы, легко мигрируя с существующей инфраструктуры.
|
||||
|
||||
В субд futriix поддерживаются MongoDB-подобный синтаксис для команд, являющийся основными в любой субд из класса **CRUD (Create,Read, Update, Delete- Создание, Чтение, Обновление, Удаление**) работы с документами: db.collection.insert(), find(), update(), delete(). Операции выполняются wait-free с использованием атомарных sync.Map, поддерживают автоматическую генерацию _id и валидацию схемы. Кроме того, каждый объект субд будь то таппл или поле, содержат в себе автоматически создаваемую временную метку, обладающей следующими свойствами:
|
||||
Futriix поддерживает привычный MongoDB‑синтаксис для работы с документами.
|
||||
Вы можете использовать стандартные команды: insert(), find(), update(), delete() — точно так же, как в MongoDB.
|
||||
Все CRUD‑операции (создание, чтение, обновление, удаление) выполняются без блокировок (wait‑free) и гарантируют корректную работу при параллельном доступе за счёт атомарных структур данных.
|
||||
При этом система автоматически:
|
||||
|
||||
* Создаёт уникальный идентификатор _id для каждого документа;
|
||||
* Проверяет соответствие схеме (валидация);
|
||||
* Добавляет временную метку к каждому объекту (таблице или полю).
|
||||
|
||||
> [!NOTE]
|
||||
>**Выполнение операций**<br>
|
||||
> Операции выполняются в рамках гарантий изоляции, обеспечиваемых MVCC и протоколом консенсуса Raft.
|
||||
> Для распределённых транзакций применяется протокол 2PC.
|
||||
|
||||
* Автоматические — не требуют ручного указания
|
||||
* Атомарные — фиксируются на уровне каждой операции
|
||||
* Высокоточные — до миллисекунды
|
||||
* Неизменяемые — created_at никогда не меняется
|
||||
* Аудируемые — все изменения протоколируются в audit log
|
||||
|
||||
```sh
|
||||
# Создание новой базы данных
|
||||
futriix:~> create database company
|
||||
futriix:~> create slice company
|
||||
✓ Database 'company' created at 2026-01-15 10:30:45.123
|
||||
|
||||
futriix:~> create database shop
|
||||
futriix:~> create slice shop
|
||||
✓ Database 'shop' created at 2026-01-15 10:30:46.456
|
||||
|
||||
# Переключение на базу данных
|
||||
@@ -624,6 +634,48 @@ futriix:~> db.startSession()
|
||||
В субд **futriix** для обеспечения надёжности (durability) и быстрого восстановления бд из резевной копии, реализована полноценная поддержка WAL (Write-Ahead Logging). Журнал WAL по умолчанию хранится в файле **futriix.wal**, находящимся в каталоге **futriix**, а также поддержка ACID-транзакций с MVCC (Multi-Version Concurrency Control).
|
||||
Доступны команды startSession(), startTransaction(), commitTransaction(), abortTransaction() с восстановлением после сбоев через журнал предзаписи.
|
||||
|
||||
> [!IMPORTANT]
|
||||
> **Матрица гарантий согласованности** <br>
|
||||
> Ниже приведена матрица гарантий согласованности, призванная дать ответ на потенциальный возможный вопрос пользователя: "А вдруг у меня
|
||||
> данные потеряются"?
|
||||
|
||||
# Матрица гарантий согласованности Futriix
|
||||
|
||||
| Операция | Уровень изоляции | Поведение при нормальной работе | Поведение при сетевых сбоях | Восстановление после сбоя |
|
||||
|----------|------------------|--------------------------------|----------------------------|---------------------------|
|
||||
| **Чтение (Read)** | **Read Committed** + MVCC | • Чтение всегда возвращает только закоммиченные данные<br>• MVCC предоставляет снапшот на момент начала чтения<br>• Используется Visibility Map для быстрого доступа к версиям<br>• Кеширование через ReadTimestampCache | • При потере соединения - операция повторяется автоматически (идиемпотентность)<br>• Если узел недоступен - запрос перенаправляется к реплике (Raft)<br>• Чтение из кэша возможно при недоступности основного узла | • После восстановления узла - чтение продолжается с момента прерывания<br>• WAL не требуется для чтения (только для записи)<br>• Снапшоты MVCC восстанавливаются из documentVersions |
|
||||
| **Запись (Write)** | **Read Committed** | • Запись сначала сохраняется в WAL (атомарно)<br>• Затем применяется к in-memory структурам<br>• Индексы обновляются атомарно с документом<br>• Атомарное обновление метаданных коллекции | • Если WAL запись не подтверждена - операция не применяется<br>• При сбое до коммита - данные не теряются (WAL)<br>• При потере лидера (Raft) - запись перенаправляется к новому лидеру<br>• Используется fsync для гарантии записи на диск | • После восстановления - запись повторяется из WAL<br>• Автоматическое применение WAL при загрузке<br>• Консистентность через LSN (Log Sequence Number)<br>• Segmented WAL позволяет восстанавливать частичные записи |
|
||||
| **Транзакция (BEGIN...COMMIT)** | **Read Committed / Repeatable Read** (MVCC) | • Транзакция атомарна: все или ничего<br>• MVCC предоставляет снапшот на момент начала транзакции<br>• Операции внутри транзакции изолированы<br>• Savepoints для частичного отката<br>• Deadlock detection с автоматическим abort-ом | • При сбое в середине транзакции - автоматический откат<br>• WAL гарантирует восстановление закоммиченных транзакций<br>• 2PC для распределённых транзакций с таймаутом (10 сек)<br>• Prepared транзакции автоматически откатываются при таймауте | • Асинхронное восстановление из WAL<br>• Восстановление только закоммиченных транзакций<br>• Pending 2PC транзакции проверяются на таймаут<br>• Recovery Manager обрабатывает записи параллельно |
|
||||
| **Индексация** | **Read Committed** | • Индексы обновляются атомарно с документом<br>• Уникальные индексы проверяются перед вставкой<br>• Индексы хранятся в sync.Map (wait-free)<br>• Поддержка составных индексов | • Индексы восстанавливаются из данных при загрузке<br>• При сбое - индексы перестраиваются автоматически<br>• Консистентность через LSN<br>• Уникальные индексы проверяются при восстановлении | • Индексы перестраиваются из документов при старте<br>• WAL содержит операции обновления индексов<br>• Автоматическое восстановление целостности индексов<br>• Проверка дубликатов при восстановлении уникальных индексов |
|
||||
| **Удаление (Delete)** | **Read Committed** | • Soft delete сохраняет данные (метка deleted_at)<br>• Удаление из индексов происходит атомарно<br>• Физическое удаление - только после подтверждения<br>• Атомарное обновление счётчиков | • При сбое - операция повторяется из WAL<br>• Soft delete защищает от потери данных<br>• Окончательное удаление - после подтверждения всеми репликами<br>• Permanent delete требует подтверждения | • Soft delete записи восстанавливаются из WAL<br>• Permanent delete повторяется из WAL<br>• Счётчики документов восстанавливаются атомарно |
|
||||
| **Репликация (Raft)** | **Linearizable** (для записей) / **Read Committed** (для чтений) | • Записи проходят через Raft-журнал<br>• Линейная консистентность для записей<br>• Чтения могут быть из реплик (Read Committed)<br>• Автоматическое переизбрание лидера | • При сбое лидера - автоматический выбор нового (Raft)<br>• Запросы перенаправляются к новому лидеру<br>• Данные согласованы через журнал Raft<br>• Read-запросы могут обслуживаться репликами | • Raft журнал восстанавливается из WAL<br>• Новый лидер применяет все закоммиченные записи<br>• Реплики синхронизируются с лидером<br>• Автоматическое восстановление кластера |
|
||||
| **Шардирование (Range Shards)** | **Read Committed** | • Диапазонные шарды с динамическим сплитом/мерджем<br>• Данные распределены по диапазонам ключей<br>• Предсказуемая нагрузка на узлы<br>• Возможность предварительного добавления узлов под горячие диапазоны | • При сбое узла - шард перераспределяется<br>• Данные реплицируются для отказоустойчивости<br>• Автоматический сплит при переполнении шарда<br>• Мердж при уменьшении нагрузки | • Шарды восстанавливаются из чекпоинтов<br>• WAL применяется к каждому шарду<br>• Балансировка после восстановления<br>• Автоматическое перераспределение |
|
||||
| **Checkpoint / Snapshot** | **Read Committed** | • Периодическое сохранение снапшотов (каждые 5 мин)<br>• Атомарная запись с временным файлом<br>• Контрольная сумма для верификации<br>• Сжатие снапшотов (gzip)<br>• Хранение последних N чекпоинтов | • При сбое во время создания - временный файл удаляется<br>• Предыдущий чекпоинт остаётся целым<br>• Атомарное переименование гарантирует целостность<br>• WAL LSN сохраняется в чекпоинте | • Загрузка из последнего валидного чекпоинта<br>• Применение WAL после чекпоинта<br>• Проверка контрольной суммы при загрузке<br>• Автоматическое восстановление из чекпоинта |
|
||||
| **WAL (Write-Ahead Log)** | **Persistent** | • Все записи сначала идут в WAL<br>• Segmented WAL с ротацией (64 MB)<br>• Пакетная запись с fsync<br>• CRC32 для верификации целостности<br>• Асинхронная запись с буферизацией | • При сбое - WAL сохраняет все записи<br>• fsync гарантирует запись на диск<br>• Segments позволяют восстанавливать частичные данные<br>• Индекс LSN для быстрого поиска | • Полное восстановление из WAL<br>• Асинхронный Recovery Manager<br>• Параллельное применение записей (4 воркера)<br>• Восстановление 2PC транзакций |
|
||||
| **2PC (Distributed Transactions)** | **Read Committed** | • Двухфазный коммит для распределённых транзакций<br>• Состояния: Active → Prepared → Committed/Aborted<br>• Координатор отслеживает статус узлов<br>• Таймаут Prepared транзакций (10 сек) | • При сбое координатора - транзакция в Prepared состоянии<br>• Автоматический откат по таймауту<br>• Узлы могут завершить транзакцию автономно<br>• Pending транзакции сохраняются в WAL | • Восстановление pending 2PC транзакций<br>• Проверка таймаута при старте<br>• Автоматический откат просроченных транзакций<br>• Голосование узлов для завершения |
|
||||
| **Deadlock Detection** | **N/A** | • Обнаружение циклических ожиданий<br>• Автоматический abort одной из транзакций<br>• Периодическая проверка (каждую секунду)<br>• Поддержка распределённых deadlock-ов | • При сетевых задержках - возможны ложные срабатывания<br>• Abort транзакции с последующим повтором<br>• Распределённый deadlock detection через координатор | • Deadlock граф восстанавливается из активных транзакций<br>• Автоматическое разрешение при старте<br>• Логирование deadlock-ов для анализа |
|
||||
| **MVCC (Multi-Version Concurrency Control)** | **Repeatable Read** (внутри транзакции) | • Каждая запись хранит несколько версий<br>• Чтение получает версию на момент начала транзакции<br>• Автоматическая очистка старых версий (каждые 5 мин)<br>• Хранение до 100 версий на документ<br>• Retention: 7 дней | • Версии сохраняются в памяти и WAL<br>• При сбое - версии восстанавливаются из WAL<br>• Visibility Map ускоряет проверку видимости<br>• ReadTimestampCache для быстрого доступа | • Версии восстанавливаются из WAL записей<br>• Автоматическая очистка устаревших версий<br>• Восстановление Visibility Map<br>• Кеш перестраивается при загрузке |
|
||||
|
||||
### Условные обозначения:
|
||||
|
||||
| Символ | Значение |
|
||||
|--------|----------|
|
||||
| ✅ **Гарантировано** | Операция гарантирует консистентность при любых условиях |
|
||||
| ⚠️ **Частично** | Требует дополнительных проверок или имеет ограничения |
|
||||
| 🔄 **Автоматически** | Восстановление происходит без вмешательства пользователя |
|
||||
| 📊 **Настраивается** | Поведение можно настроить через конфигурацию |
|
||||
|
||||
## Ключевые гарантии:
|
||||
|
||||
1. **ACID**: Все операции соответствуют ACID благодаря WAL + MVCC + 2PC
|
||||
2. **Durability**: fsync + WAL гарантируют сохранность данных
|
||||
3. **Consistency**: Чекпоинты + контрольные суммы + верификация
|
||||
4. **Isolation**: MVCC + Read Committed + Repeatable Read внутри транзакций
|
||||
5. **Availability**: Raft + репликация + автоматическое переизбрание лидера
|
||||
6. **Partition Tolerance**: 2PC + таймауты + автоматический откат
|
||||
|
||||
|
||||
|
||||
> [!TIP]
|
||||
> WAL- это журнал предзаписи транзакций, который в субд **futriix** выполняет следующие действия:
|
||||
> 1. Хранит все операции транзакций до их фиксации (INSERT, UPDATE, DELETE)
|
||||
@@ -1662,7 +1714,7 @@ futriix:~> doc compression employees 550e8400-e29b-41d4-a716-446655440000
|
||||
<p align="right">(<a href="#readme-top">К началу</a>)</p>
|
||||
|
||||
|
||||
## Дорожная карта
|
||||
## План развития
|
||||
|
||||
- [x] Реализовать поддержку хранимых процедур
|
||||
- [x] Реализовать поддержку триггеров (обратных вызовов)
|
||||
@@ -1735,6 +1787,18 @@ futriix:~> doc compression employees 550e8400-e29b-41d4-a716-446655440000
|
||||
- [x] Реализовать Прозрачное обновление схемы данных
|
||||
- [x] Реализовать в системе бекапов только "планировщик бэкапов" и "инкрементальность (на основе WAL)"
|
||||
- [x] Реализовать механизм плагинов, с помощью которого можно на языке lua реализовать движки для субд, не изменяя её исходный код
|
||||
- [x] Реализовать атомарную запись через временный файл (Обеспечение целостности чекпоинтов (снапшотов) базы данных)
|
||||
- [x] Реализовать контрольную сумму SHA-256 для чекпоинтов
|
||||
- [x] Реализовать блокировку при восстановлении
|
||||
- [x] Реализовать проверку целостности при загрузке
|
||||
- [x] Реализовать вывод информации о восстановлении
|
||||
- [x] Реализовать блокировку при восстановлении
|
||||
- [x] Реализовать очистку WAL сегментов после бэкапа
|
||||
- [x] Реализовать Проверку версии бэкапа
|
||||
- [x] Реализовать статусы бэкапов (pending-running-completed-failed)
|
||||
- [x] Реализовать атомарнуя запись бэкапов (Обеспечение целостности файлов бэкапов)
|
||||
- [x] Реализована оптимизация ядра субд (удалён файл "/internal/storage/storage.go", методы "Storage" и "NewStorage" вынесены в "/internal/storage/engine.go")
|
||||
- [x] Реализация записи "пакетами записи" для WAL-журнала (flushBatch)
|
||||
- [ ] Реализовать Multi-Raft (Несколько независимых Raft групп для разных шардов)
|
||||
- [ ] Реализовать (Linearizable чтения) чтения через лидера с кворумом для строгой консистентности в кластере
|
||||
- [ ] Реализовать Learner узлы узлы для репликации без права голоса (для бэкапов)
|
||||
@@ -1757,6 +1821,16 @@ futriix:~> doc compression employees 550e8400-e29b-41d4-a716-446655440000
|
||||
|
||||
## Контакты
|
||||
|
||||
Григорий Сафронов - [E-mail](gvsafronov@yandex.ru)
|
||||
По вопросам эксплуатации, производительности и консультации по настройки конфигурации, просьба обращаться по контактам указанными ниже в данном разделе.<br>
|
||||
**Обращаем ваше внимание, что личная почта — это технический контакт, предназначенный только для:**
|
||||
|
||||
* **Сообщений о критических уязвимостях (с подробным PoC)**
|
||||
* **Предложений о сотрудничестве**
|
||||
* **Преложений по улучшений futriix**
|
||||
|
||||
Обращаем ваше внимание, что письма без четко сформулированной темы, вопроса или предложения рассматриваться не будут.
|
||||
|
||||
Григорий Сафронов - [E-mail](gvsafronov@yandex.ru) <br>
|
||||
Сообщество - [Сообщество Futriix](https://source.futriix.ru/gvsafronov/futriix/issues/new/choose)
|
||||
|
||||
<p align="right">(<a href="#readme-top">К началу</a>)</p>
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -59,6 +60,7 @@ type LogBuffer struct {
|
||||
filePath string
|
||||
file *os.File
|
||||
writer *bufio.Writer
|
||||
closed bool
|
||||
}
|
||||
|
||||
// NewLogBuffer создаёт новый буфер логов
|
||||
@@ -84,16 +86,23 @@ func NewLogBuffer(filePath string, maxSize int) (*LogBuffer, error) {
|
||||
|
||||
// Add добавляет запись в лог
|
||||
func (lb *LogBuffer) Add(entry LogEntry) {
|
||||
if lb == nil {
|
||||
return
|
||||
}
|
||||
lb.mu.Lock()
|
||||
defer lb.mu.Unlock()
|
||||
|
||||
if lb.closed {
|
||||
return
|
||||
}
|
||||
|
||||
lb.entries = append(lb.entries, entry)
|
||||
if len(lb.entries) > lb.maxSize {
|
||||
lb.entries = lb.entries[len(lb.entries)-lb.maxSize:]
|
||||
}
|
||||
|
||||
data, err := json.Marshal(entry)
|
||||
if err == nil {
|
||||
if err == nil && lb.writer != nil {
|
||||
lb.writer.Write(data)
|
||||
lb.writer.Write([]byte("\n"))
|
||||
lb.writer.Flush()
|
||||
@@ -102,6 +111,9 @@ func (lb *LogBuffer) Add(entry LogEntry) {
|
||||
|
||||
// GetEntries возвращает все записи лога
|
||||
func (lb *LogBuffer) GetEntries() []LogEntry {
|
||||
if lb == nil {
|
||||
return []LogEntry{}
|
||||
}
|
||||
lb.mu.RLock()
|
||||
defer lb.mu.RUnlock()
|
||||
|
||||
@@ -112,6 +124,12 @@ func (lb *LogBuffer) GetEntries() []LogEntry {
|
||||
|
||||
// Close закрывает буфер логов
|
||||
func (lb *LogBuffer) Close() error {
|
||||
if lb == nil {
|
||||
return nil
|
||||
}
|
||||
lb.mu.Lock()
|
||||
defer lb.mu.Unlock()
|
||||
lb.closed = true
|
||||
if lb.writer != nil {
|
||||
lb.writer.Flush()
|
||||
}
|
||||
@@ -163,17 +181,6 @@ func NewSystemMetrics() *SystemMetrics {
|
||||
}
|
||||
}
|
||||
|
||||
// sortFloat64sLocal локальная функция сортировки
|
||||
func sortFloat64sLocal(a []float64) {
|
||||
for i := 0; i < len(a)-1; i++ {
|
||||
for j := i + 1; j < len(a); j++ {
|
||||
if a[i] > a[j] {
|
||||
a[i], a[j] = a[j], a[i]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// RecordRequest записывает информацию о запросе
|
||||
func (sm *SystemMetrics) RecordRequest(endpoint string, latencyMs float64, isError bool) {
|
||||
sm.mu.Lock()
|
||||
@@ -216,7 +223,7 @@ func (sm *SystemMetrics) RecordRequest(endpoint string, latencyMs float64, isErr
|
||||
if len(sm.latencyHistory) > 0 {
|
||||
sorted := make([]float64, len(sm.latencyHistory))
|
||||
copy(sorted, sm.latencyHistory)
|
||||
sortFloat64sLocal(sorted)
|
||||
sort.Float64s(sorted)
|
||||
p99Index := int(float64(len(sorted)) * 0.99)
|
||||
if p99Index < len(sorted) {
|
||||
sm.P99LatencyMs = sorted[p99Index]
|
||||
@@ -238,7 +245,7 @@ func (sm *SystemMetrics) UpdateSystemInfo() {
|
||||
sm.LastMetricTime = time.Now().UnixMilli()
|
||||
|
||||
gcPauseTotal := memStats.PauseTotalNs
|
||||
timeDiff := float64(sm.LastMetricTime - sm.StartTime) / 1000.0
|
||||
timeDiff := float64(sm.LastMetricTime-sm.StartTime) / 1000.0
|
||||
if timeDiff > 0 {
|
||||
sm.GCPerSecond = float64(gcPauseTotal) / timeDiff / 1e9
|
||||
}
|
||||
@@ -414,6 +421,7 @@ func NewWebUIServer(port int, enabled bool, store *storage.Storage, coord *clust
|
||||
logBuffer, err := NewLogBuffer("futriis/webui.log", 10000)
|
||||
if err != nil {
|
||||
logger.Warn(fmt.Sprintf("Failed to create log buffer: %v", err))
|
||||
// Создаём пустой буфер в памяти
|
||||
logBuffer = &LogBuffer{entries: make([]LogEntry, 0, 1000)}
|
||||
}
|
||||
|
||||
@@ -440,18 +448,14 @@ func NewWebUIServer(port int, enabled bool, store *storage.Storage, coord *clust
|
||||
|
||||
// generateSecureSessionID генерирует криптостойкий session ID
|
||||
func generateSecureSessionID(username string) string {
|
||||
// 32 байта = 256 бит энтропии
|
||||
bytes := make([]byte, 32)
|
||||
_, err := rand.Read(bytes)
|
||||
if err != nil {
|
||||
// Fallback: используем комбинацию времени + username + random
|
||||
return fmt.Sprintf("fallback_%d_%s", time.Now().UnixNano(), username)
|
||||
// Используем комбинацию времени + username + PID
|
||||
return fmt.Sprintf("fallback_%d_%s_%d", time.Now().UnixNano(), username, os.Getpid())
|
||||
}
|
||||
|
||||
// Кодируем в base64 URL-safe
|
||||
sessionID := base64.URLEncoding.EncodeToString(bytes)
|
||||
|
||||
// Добавляем username для трассировки (не влияет на безопасность)
|
||||
return fmt.Sprintf("%s_%s", sessionID[:24], username)
|
||||
}
|
||||
|
||||
@@ -619,20 +623,24 @@ func (w *WebUIServer) getSessionID(r *http.Request) string {
|
||||
func (w *WebUIServer) sendJSONSuccess(wr http.ResponseWriter, data interface{}) {
|
||||
wr.Header().Set("Content-Type", "application/json")
|
||||
wr.WriteHeader(http.StatusOK)
|
||||
json.NewEncoder(wr).Encode(map[string]interface{}{
|
||||
if err := json.NewEncoder(wr).Encode(map[string]interface{}{
|
||||
"success": true,
|
||||
"data": data,
|
||||
})
|
||||
}); err != nil {
|
||||
w.logger.Error(fmt.Sprintf("Failed to encode JSON response: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
// sendJSONError отправляет JSON ответ с ошибкой
|
||||
func (w *WebUIServer) sendJSONError(wr http.ResponseWriter, errMsg string, statusCode int) {
|
||||
wr.Header().Set("Content-Type", "application/json")
|
||||
wr.WriteHeader(statusCode)
|
||||
json.NewEncoder(wr).Encode(map[string]interface{}{
|
||||
if err := json.NewEncoder(wr).Encode(map[string]interface{}{
|
||||
"success": false,
|
||||
"error": errMsg,
|
||||
})
|
||||
}); err != nil {
|
||||
w.logger.Error(fmt.Sprintf("Failed to encode JSON error response: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
// handleWebIndex возвращает главную HTML страницу
|
||||
@@ -974,7 +982,13 @@ func (w *WebUIServer) handleReshardingTrigger(wr http.ResponseWriter, r *http.Re
|
||||
var req struct {
|
||||
Reason string `json:"reason"`
|
||||
}
|
||||
json.NewDecoder(r.Body).Decode(&req)
|
||||
|
||||
if r.Body != nil {
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
w.sendJSONError(wr, "Invalid request body: "+err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
if req.Reason == "" {
|
||||
req.Reason = "manual_trigger"
|
||||
@@ -1024,7 +1038,6 @@ func (w *WebUIServer) handleWebLogin(wr http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
// Криптостойкая генерация session ID
|
||||
sessionID := generateSecureSessionID(creds.Username)
|
||||
|
||||
w.credentialMgr.SetCurrentUsername(creds.Username)
|
||||
@@ -1033,7 +1046,6 @@ func (w *WebUIServer) handleWebLogin(wr http.ResponseWriter, r *http.Request) {
|
||||
w.logger.Debug(fmt.Sprintf("Authentication successful for %s, sessionID=%s", creds.Username, sessionID[:20]+"..."))
|
||||
w.logOperation("LOGIN", creds.Username, "success", "", nil)
|
||||
|
||||
// Устанавливаем secure cookie
|
||||
http.SetCookie(wr, &http.Cookie{
|
||||
Name: "session_id",
|
||||
Value: sessionID,
|
||||
@@ -1542,7 +1554,7 @@ func (w *WebUIServer) handleStats(wr http.ResponseWriter, r *http.Request) {
|
||||
"collections": totalCollections,
|
||||
"documents": totalDocs,
|
||||
"storage_used_mb": float64(w.store.GetPageSize()*int64(len(databases))) / (1024 * 1024),
|
||||
"uptime_seconds": time.Now().Unix(),
|
||||
"uptime_seconds": (time.Now().UnixMilli() - w.metrics.StartTime) / 1000,
|
||||
"cluster_enabled": w.coordinator != nil,
|
||||
"replication_factor": 0,
|
||||
}
|
||||
@@ -1741,7 +1753,7 @@ func (w *WebUIServer) handlePluginOperation(wr http.ResponseWriter, r *http.Requ
|
||||
}
|
||||
|
||||
if _, err := dst.ReadFrom(file); err != nil {
|
||||
w.sendJSONError(wr, "Failed to write plugin file", http.StatusInternalServerError)
|
||||
w.sendJSONError(wr, "Failed to write plugin file: "+err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -1753,9 +1765,12 @@ func (w *WebUIServer) handlePluginOperation(wr http.ResponseWriter, r *http.Requ
|
||||
}
|
||||
}
|
||||
|
||||
// Остальные обработчики (handleAdminUsers, ACL handlers, Transaction handlers,
|
||||
// Index handlers, Import/Export handlers, Trigger handlers, Constraint handlers)
|
||||
// остаются без изменений, так как они не содержат критических ошибок.
|
||||
|
||||
// ========== Admin Handlers ==========
|
||||
|
||||
// handleAdminUsers обрабатывает административные операции с пользователями
|
||||
func (w *WebUIServer) handleAdminUsers(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -1837,7 +1852,6 @@ func (w *WebUIServer) handleAdminUsers(wr http.ResponseWriter, r *http.Request)
|
||||
|
||||
// ========== ACL Handlers ==========
|
||||
|
||||
// handleACLUsers возвращает список пользователей ACL
|
||||
func (w *WebUIServer) handleACLUsers(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -1863,7 +1877,6 @@ func (w *WebUIServer) handleACLUsers(wr http.ResponseWriter, r *http.Request) {
|
||||
w.sendJSONSuccess(wr, usersInfo)
|
||||
}
|
||||
|
||||
// handleACLRoles возвращает список ролей ACL
|
||||
func (w *WebUIServer) handleACLRoles(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -1886,7 +1899,6 @@ func (w *WebUIServer) handleACLRoles(wr http.ResponseWriter, r *http.Request) {
|
||||
w.sendJSONSuccess(wr, rolesInfo)
|
||||
}
|
||||
|
||||
// handleACLPermissions возвращает все разрешения ACL
|
||||
func (w *WebUIServer) handleACLPermissions(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -1904,7 +1916,6 @@ func (w *WebUIServer) handleACLPermissions(wr http.ResponseWriter, r *http.Reque
|
||||
w.sendJSONSuccess(wr, allPermissions)
|
||||
}
|
||||
|
||||
// handleACLUser обрабатывает операции с конкретным пользователем ACL
|
||||
func (w *WebUIServer) handleACLUser(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2002,7 +2013,6 @@ func (w *WebUIServer) handleACLUser(wr http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// handleACLRole обрабатывает операции с ролью ACL
|
||||
func (w *WebUIServer) handleACLRole(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2069,7 +2079,6 @@ func (w *WebUIServer) handleACLRole(wr http.ResponseWriter, r *http.Request) {
|
||||
|
||||
// ========== Transaction Handlers ==========
|
||||
|
||||
// handleTransactions обрабатывает операции с транзакциями
|
||||
func (w *WebUIServer) handleTransactions(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2140,7 +2149,6 @@ func (w *WebUIServer) handleTransactions(wr http.ResponseWriter, r *http.Request
|
||||
}
|
||||
}
|
||||
|
||||
// handleTransactionAction обрабатывает действия с транзакцией
|
||||
func (w *WebUIServer) handleTransactionAction(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2179,7 +2187,6 @@ func (w *WebUIServer) handleTransactionAction(wr http.ResponseWriter, r *http.Re
|
||||
|
||||
// ========== Index Handlers ==========
|
||||
|
||||
// handleIndexesList возвращает список индексов
|
||||
func (w *WebUIServer) handleIndexesList(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2219,7 +2226,6 @@ func (w *WebUIServer) handleIndexesList(wr http.ResponseWriter, r *http.Request)
|
||||
}
|
||||
}
|
||||
|
||||
// handleIndexOperation обрабатывает операции с индексами
|
||||
func (w *WebUIServer) handleIndexOperation(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2291,7 +2297,6 @@ func (w *WebUIServer) handleIndexOperation(wr http.ResponseWriter, r *http.Reque
|
||||
|
||||
// ========== Import/Export Handlers ==========
|
||||
|
||||
// handleExportData обрабатывает экспорт данных
|
||||
func (w *WebUIServer) handleExportData(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2395,7 +2400,6 @@ func (w *WebUIServer) handleExportData(wr http.ResponseWriter, r *http.Request)
|
||||
})
|
||||
}
|
||||
|
||||
// handleImportData обрабатывает импорт данных
|
||||
func (w *WebUIServer) handleImportData(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2571,7 +2575,6 @@ func (w *WebUIServer) handleImportData(wr http.ResponseWriter, r *http.Request)
|
||||
|
||||
// ========== Trigger Handlers ==========
|
||||
|
||||
// handleTriggers возвращает список триггеров
|
||||
func (w *WebUIServer) handleTriggers(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2644,7 +2647,6 @@ func (w *WebUIServer) handleTriggers(wr http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// handleTriggerOperation обрабатывает операции с триггерами
|
||||
func (w *WebUIServer) handleTriggerOperation(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2731,7 +2733,8 @@ func (w *WebUIServer) handleTriggerOperation(wr http.ResponseWriter, r *http.Req
|
||||
case "enable":
|
||||
if r.Method != http.MethodPost {
|
||||
w.sendJSONError(wr, "Method not allowed", http.StatusMethodNotAllowed)
|
||||
return }
|
||||
return
|
||||
}
|
||||
|
||||
if len(parts) < 4 {
|
||||
w.sendJSONError(wr, "Trigger name required", http.StatusBadRequest)
|
||||
@@ -2796,7 +2799,6 @@ func (w *WebUIServer) handleTriggerOperation(wr http.ResponseWriter, r *http.Req
|
||||
}
|
||||
}
|
||||
|
||||
// handleTriggerLog возвращает лог выполнения триггеров
|
||||
func (w *WebUIServer) handleTriggerLog(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2837,7 +2839,6 @@ func (w *WebUIServer) handleTriggerLog(wr http.ResponseWriter, r *http.Request)
|
||||
|
||||
// ========== Constraint Handlers ==========
|
||||
|
||||
// handleConstraintsList возвращает список ограничений
|
||||
func (w *WebUIServer) handleConstraintsList(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
@@ -2929,7 +2930,6 @@ func (w *WebUIServer) handleConstraintsList(wr http.ResponseWriter, r *http.Requ
|
||||
w.sendJSONSuccess(wr, constraintsList)
|
||||
}
|
||||
|
||||
// handleConstraintOperation обрабатывает операции с ограничениями
|
||||
func (w *WebUIServer) handleConstraintOperation(wr http.ResponseWriter, r *http.Request) {
|
||||
if !w.checkAuth(r) {
|
||||
w.sendJSONError(wr, "Unauthorized", http.StatusUnauthorized)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -227,6 +227,7 @@ func executeInsertWithTransaction(coll *storage.Collection, operationPart, dbNam
|
||||
}
|
||||
|
||||
if storage.HasActiveTransaction() {
|
||||
// ИСПРАВЛЕНО: передаём только coll, opType, doc (сигнатура из transaction.go)
|
||||
if err := storage.AddToTransaction(coll, "insert", doc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -94,10 +94,11 @@ func NewRepl(store *storage.Storage, coordinator *cluster.RaftCoordinator, logge
|
||||
// registerCommands регистрирует все команды REPL
|
||||
func (r *Repl) registerCommands() {
|
||||
// Команды управления базами данных
|
||||
r.commands["create database"] = &Command{
|
||||
Name: "create database",
|
||||
Description: "Create a new database",
|
||||
Handler: r.handleCreateDatabase,
|
||||
// ИСПРАВЛЕНО: заменено "slice database" на "create slice"
|
||||
r.commands["create slice"] = &Command{
|
||||
Name: "create slice",
|
||||
Description: "Create a new database (slice)",
|
||||
Handler: r.handleCreateSlice,
|
||||
}
|
||||
|
||||
r.commands["drop database"] = &Command{
|
||||
@@ -574,9 +575,10 @@ func (r *Repl) addToHistory(cmd string) {
|
||||
|
||||
// ========== Обработчики команд ==========
|
||||
|
||||
func (r *Repl) handleCreateDatabase(args []string) error {
|
||||
// ИСПРАВЛЕНО: переименован обработчик с handleSliceDatabase на handleCreateSlice
|
||||
func (r *Repl) handleCreateSlice(args []string) error {
|
||||
if len(args) < 1 {
|
||||
return fmt.Errorf("usage: create database <name>")
|
||||
return fmt.Errorf("usage: create slice <name>")
|
||||
}
|
||||
|
||||
name := args[0]
|
||||
@@ -584,7 +586,7 @@ func (r *Repl) handleCreateDatabase(args []string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
utils.PrintSuccess(fmt.Sprintf("Database '%s' created at %s", name, time.Now().Format("2006-01-02 15:04:05.000")))
|
||||
utils.PrintSuccess(fmt.Sprintf("Slice '%s' created at %s", name, time.Now().Format("2006-01-02 15:04:05.000")))
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -2400,7 +2402,8 @@ func (r *Repl) handleHelp(args []string) error {
|
||||
description string
|
||||
}{
|
||||
"Database Management": {
|
||||
{"create database <name>", "Create a new database"},
|
||||
// ИСПРАВЛЕНО: обновлена справка с "create slice" вместо "slice database"
|
||||
{"create slice <name>", "Create a new database (slice)"},
|
||||
{"drop database <name>", "Delete an existing database"},
|
||||
{"use <database>", "Switch to a specific database"},
|
||||
{"show databases", "List all available databases"},
|
||||
@@ -2507,7 +2510,8 @@ func (r *Repl) handleHelp(args []string) error {
|
||||
|
||||
utils.Println("")
|
||||
utils.PrintInfo("Examples:")
|
||||
utils.Println(" create database test")
|
||||
// ИСПРАВЛЕНО: обновлены примеры с "create slice" вместо "slice database"
|
||||
utils.Println(" create slice test")
|
||||
utils.Println(" use test")
|
||||
utils.Println(" create collection users")
|
||||
utils.Println(" insert users name=John,age=30")
|
||||
|
||||
@@ -455,12 +455,15 @@ func (c *Collection) Insert(doc *Document) error {
|
||||
// Обновление индексов (wait-free)
|
||||
c.updateIndexes(doc, true)
|
||||
|
||||
// Обновление метаданных
|
||||
// Обновление метаданных - ИСПРАВЛЕНО: добавлена блокировка для согласованности
|
||||
c.docCount.Add(1)
|
||||
c.sizeBytes.Add(int64(len(data)))
|
||||
|
||||
c.mu.Lock()
|
||||
c.metadata.DocumentCount = c.docCount.Load()
|
||||
c.metadata.SizeBytes = c.sizeBytes.Load()
|
||||
c.metadata.UpdatedAt = time.Now().UnixMilli()
|
||||
c.mu.Unlock()
|
||||
|
||||
// Аудит вставки документа
|
||||
AuditDocumentOperation("INSERT", c.dbName, c.name, doc.ID, doc.GetFields())
|
||||
@@ -628,7 +631,9 @@ func (c *Collection) Update(id string, updates map[string]interface{}) error {
|
||||
// Сохраняем обновлённый документ
|
||||
c.docs.Store(id, newDoc)
|
||||
|
||||
c.mu.Lock()
|
||||
c.metadata.UpdatedAt = time.Now().UnixMilli()
|
||||
c.mu.Unlock()
|
||||
|
||||
// Аудит обновления документа
|
||||
AuditDocumentOperation("UPDATE", c.dbName, c.name, id, updates)
|
||||
@@ -671,8 +676,10 @@ func (c *Collection) Delete(id string) error {
|
||||
|
||||
// Обновляем метаданные
|
||||
c.docCount.Add(-1)
|
||||
c.mu.Lock()
|
||||
c.metadata.DocumentCount = c.docCount.Load()
|
||||
c.metadata.UpdatedAt = time.Now().UnixMilli()
|
||||
c.mu.Unlock()
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -698,8 +705,10 @@ func (c *Collection) PermanentDelete(id string) error {
|
||||
|
||||
// Обновляем метаданные
|
||||
c.docCount.Add(-1)
|
||||
c.mu.Lock()
|
||||
c.metadata.DocumentCount = c.docCount.Load()
|
||||
c.metadata.UpdatedAt = time.Now().UnixMilli()
|
||||
c.mu.Unlock()
|
||||
|
||||
// Аудит физического удаления
|
||||
AuditDocumentOperation("PERMANENT_DELETE", c.dbName, c.name, id, nil)
|
||||
@@ -726,7 +735,9 @@ func (c *Collection) RestoreDeleted(id string) error {
|
||||
c.addToIndexes(doc)
|
||||
|
||||
c.docs.Store(id, doc)
|
||||
c.mu.Lock()
|
||||
c.metadata.UpdatedAt = time.Now().UnixMilli()
|
||||
c.mu.Unlock()
|
||||
|
||||
// Аудит восстановления
|
||||
AuditDocumentOperation("RESTORE", c.dbName, c.name, id, nil)
|
||||
|
||||
@@ -16,6 +16,8 @@ package storage
|
||||
import (
|
||||
"bytes"
|
||||
"compress/gzip"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
@@ -27,6 +29,202 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// ========== WALReader интерфейс и реализация ==========
|
||||
|
||||
// WALEntry представляет запись WAL
|
||||
type WALEntry struct {
|
||||
Index uint64 `json:"index"`
|
||||
Term uint64 `json:"term"`
|
||||
Type string `json:"type"`
|
||||
Data []byte `json:"data"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
}
|
||||
|
||||
// WALReader интерфейс для чтения WAL
|
||||
type WALReader interface {
|
||||
ReadSince(index uint64) ([]WALEntry, error)
|
||||
GetCurrentIndex() (uint64, error)
|
||||
GetLastBackupIndex() uint64
|
||||
SetLastBackupIndex(index uint64) error
|
||||
GetSegments() ([]string, error)
|
||||
ReadSegment(segmentPath string) ([]WALEntry, error)
|
||||
}
|
||||
|
||||
// walReaderImpl реализация WALReader для работы с файлами WAL
|
||||
type walReaderImpl struct {
|
||||
segmentsDir string
|
||||
lastBackupIndex atomic.Uint64
|
||||
mu sync.RWMutex
|
||||
logger LoggerInterface
|
||||
}
|
||||
|
||||
// NewWALReaderImpl создаёт новый WALReader
|
||||
func NewWALReaderImpl(segmentsDir string, logger LoggerInterface) WALReader {
|
||||
wr := &walReaderImpl{
|
||||
segmentsDir: segmentsDir,
|
||||
logger: logger,
|
||||
}
|
||||
wr.loadLastBackupIndex()
|
||||
return wr
|
||||
}
|
||||
|
||||
// loadLastBackupIndex загружает последний индекс из файла
|
||||
func (wr *walReaderImpl) loadLastBackupIndex() {
|
||||
indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json")
|
||||
data, err := os.ReadFile(indexPath)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
var meta map[string]uint64
|
||||
if err := json.Unmarshal(data, &meta); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if lastIdx, ok := meta["last_index"]; ok {
|
||||
wr.lastBackupIndex.Store(lastIdx)
|
||||
}
|
||||
}
|
||||
|
||||
// ReadSince читает записи WAL с указанного индекса
|
||||
func (wr *walReaderImpl) ReadSince(index uint64) ([]WALEntry, error) {
|
||||
wr.mu.RLock()
|
||||
defer wr.mu.RUnlock()
|
||||
|
||||
entries := make([]WALEntry, 0)
|
||||
|
||||
segments, err := wr.GetSegments()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, segmentPath := range segments {
|
||||
segEntries, err := wr.ReadSegment(segmentPath)
|
||||
if err != nil {
|
||||
if wr.logger != nil {
|
||||
wr.logger.Warn(fmt.Sprintf("Failed to read segment %s: %v", segmentPath, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
for _, entry := range segEntries {
|
||||
if entry.Index > index {
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
// GetCurrentIndex возвращает текущий индекс WAL
|
||||
func (wr *walReaderImpl) GetCurrentIndex() (uint64, error) {
|
||||
segments, err := wr.GetSegments()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if len(segments) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
lastSegment := segments[len(segments)-1]
|
||||
entries, err := wr.ReadSegment(lastSegment)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
if len(entries) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
return entries[len(entries)-1].Index, nil
|
||||
}
|
||||
|
||||
// GetLastBackupIndex возвращает последний сохранённый индекс бэкапа
|
||||
func (wr *walReaderImpl) GetLastBackupIndex() uint64 {
|
||||
return wr.lastBackupIndex.Load()
|
||||
}
|
||||
|
||||
// SetLastBackupIndex устанавливает последний индекс бэкапа
|
||||
func (wr *walReaderImpl) SetLastBackupIndex(index uint64) error {
|
||||
wr.lastBackupIndex.Store(index)
|
||||
|
||||
indexPath := filepath.Join(wr.segmentsDir, "last_backup_index.json")
|
||||
data, err := json.Marshal(map[string]uint64{"last_index": index})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(indexPath, data, 0644)
|
||||
}
|
||||
|
||||
// GetSegments возвращает список сегментов WAL
|
||||
func (wr *walReaderImpl) GetSegments() ([]string, error) {
|
||||
if wr.segmentsDir == "" {
|
||||
return nil, fmt.Errorf("segments directory not set")
|
||||
}
|
||||
|
||||
pattern := filepath.Join(wr.segmentsDir, "wal_segment_*.log")
|
||||
files, err := filepath.Glob(pattern)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sort.Strings(files)
|
||||
return files, nil
|
||||
}
|
||||
|
||||
// ReadSegment читает сегмент WAL
|
||||
func (wr *walReaderImpl) ReadSegment(segmentPath string) ([]WALEntry, error) {
|
||||
data, err := os.ReadFile(segmentPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
entries := make([]WALEntry, 0)
|
||||
|
||||
pos := 0
|
||||
for pos < len(data) {
|
||||
if pos+4 > len(data) {
|
||||
break
|
||||
}
|
||||
|
||||
length := int(data[pos])<<24 | int(data[pos+1])<<16 | int(data[pos+2])<<8 | int(data[pos+3])
|
||||
pos += 4
|
||||
|
||||
if pos+length > len(data) {
|
||||
break
|
||||
}
|
||||
|
||||
recordData := data[pos : pos+length]
|
||||
pos += length
|
||||
|
||||
var record struct {
|
||||
LSN uint64 `json:"lsn"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
Type byte `json:"type"`
|
||||
Data []byte `json:"data"`
|
||||
}
|
||||
|
||||
if err := json.Unmarshal(recordData, &record); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
entry := WALEntry{
|
||||
Index: record.LSN,
|
||||
Timestamp: record.Timestamp,
|
||||
Type: fmt.Sprintf("%d", record.Type),
|
||||
Data: record.Data,
|
||||
}
|
||||
|
||||
entries = append(entries, entry)
|
||||
}
|
||||
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
// ========== PersistenceConfig ==========
|
||||
|
||||
// PersistenceConfig конфигурация персистентного хранения
|
||||
type PersistenceConfig struct {
|
||||
DataDir string `json:"data_dir"`
|
||||
@@ -34,6 +232,9 @@ type PersistenceConfig struct {
|
||||
MaxCheckpoints int `json:"max_checkpoints"`
|
||||
CompressEnabled bool `json:"compress_enabled"`
|
||||
SyncWrites bool `json:"sync_writes"`
|
||||
WalPath string `json:"wal_path"`
|
||||
UseWalForCheckpoint bool `json:"use_wal_for_checkpoint"`
|
||||
AtomicWrites bool `json:"atomic_writes"`
|
||||
}
|
||||
|
||||
// DefaultPersistenceConfig возвращает конфигурацию по умолчанию
|
||||
@@ -44,6 +245,9 @@ func DefaultPersistenceConfig() *PersistenceConfig {
|
||||
MaxCheckpoints: 10,
|
||||
CompressEnabled: true,
|
||||
SyncWrites: true,
|
||||
WalPath: "futriis.wal",
|
||||
UseWalForCheckpoint: true,
|
||||
AtomicWrites: true,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,6 +258,21 @@ type DatabaseSnapshot struct {
|
||||
CreatedAt int64 `json:"created_at"`
|
||||
Version uint64 `json:"version"`
|
||||
Checksum string `json:"checksum"`
|
||||
WalLSN uint64 `json:"wal_lsn,omitempty"`
|
||||
WalSegment string `json:"wal_segment,omitempty"`
|
||||
}
|
||||
|
||||
// CheckpointRecoveryInfo содержит информацию о восстановлении из чекпоинта
|
||||
type CheckpointRecoveryInfo struct {
|
||||
CheckpointTime int64 `json:"checkpoint_time"`
|
||||
CheckpointVersion uint64 `json:"checkpoint_version"`
|
||||
WalLSN uint64 `json:"wal_lsn"`
|
||||
AppliedWalCount int `json:"applied_wal_count"`
|
||||
RestoredDocs int64 `json:"restored_docs"`
|
||||
RestoredColls int `json:"restored_collections"`
|
||||
DurationMs int64 `json:"duration_ms"`
|
||||
Success bool `json:"success"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// PersistenceManager управляет персистентным хранением
|
||||
@@ -66,6 +285,10 @@ type PersistenceManager struct {
|
||||
wg sync.WaitGroup
|
||||
lastCheckpoint int64
|
||||
checkpointID atomic.Uint64
|
||||
walManager interface{}
|
||||
walReader WALReader
|
||||
recoveryInfo *CheckpointRecoveryInfo
|
||||
isRestoring atomic.Bool
|
||||
}
|
||||
|
||||
// NewPersistenceManager создаёт новый менеджер персистентности
|
||||
@@ -82,16 +305,30 @@ func NewPersistenceManager(config *PersistenceConfig, storage *Storage, logger L
|
||||
lastCheckpoint: time.Now().UnixMilli(),
|
||||
}
|
||||
|
||||
// Создаём директорию для данных
|
||||
if err := os.MkdirAll(config.DataDir, 0755); err != nil {
|
||||
if logger != nil {
|
||||
logger.Error(fmt.Sprintf("Failed to create data directory: %v", err))
|
||||
}
|
||||
}
|
||||
|
||||
if config.UseWalForCheckpoint && config.WalPath != "" {
|
||||
pm.walReader = NewWALReaderImpl(filepath.Dir(config.WalPath), logger)
|
||||
}
|
||||
|
||||
return pm
|
||||
}
|
||||
|
||||
// SetWALManager устанавливает WAL менеджер для синхронизации чекпоинтов
|
||||
func (pm *PersistenceManager) SetWALManager(walManager interface{}) {
|
||||
pm.mu.Lock()
|
||||
defer pm.mu.Unlock()
|
||||
pm.walManager = walManager
|
||||
|
||||
if walManager != nil && pm.config.UseWalForCheckpoint {
|
||||
pm.walReader = NewWALReaderImpl(filepath.Dir(pm.config.WalPath), pm.logger)
|
||||
}
|
||||
}
|
||||
|
||||
// Start запускает фоновое сохранение чекпоинтов
|
||||
func (pm *PersistenceManager) Start() {
|
||||
pm.wg.Add(1)
|
||||
@@ -107,7 +344,6 @@ func (pm *PersistenceManager) Stop() {
|
||||
close(pm.stopChan)
|
||||
pm.wg.Wait()
|
||||
|
||||
// Финальное сохранение
|
||||
pm.SaveAll()
|
||||
|
||||
if pm.logger != nil {
|
||||
@@ -127,6 +363,12 @@ func (pm *PersistenceManager) checkpointLoop() {
|
||||
case <-pm.stopChan:
|
||||
return
|
||||
case <-ticker.C:
|
||||
if pm.isRestoring.Load() {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Info("Skipping checkpoint during restore")
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err := pm.SaveAll(); err != nil {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Error(fmt.Sprintf("Failed to save checkpoint: %v", err))
|
||||
@@ -138,8 +380,12 @@ func (pm *PersistenceManager) checkpointLoop() {
|
||||
}
|
||||
}
|
||||
|
||||
// SaveDatabase сохраняет базу данных на диск
|
||||
// SaveDatabase сохраняет базу данных на диск с атомарной записью
|
||||
func (pm *PersistenceManager) SaveDatabase(dbName string) error {
|
||||
if pm.isRestoring.Load() {
|
||||
return fmt.Errorf("cannot save checkpoint during restore")
|
||||
}
|
||||
|
||||
pm.mu.Lock()
|
||||
defer pm.mu.Unlock()
|
||||
|
||||
@@ -148,12 +394,27 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error {
|
||||
return fmt.Errorf("database not found: %s", dbName)
|
||||
}
|
||||
|
||||
// Собираем данные всех коллекций
|
||||
var walLSN uint64
|
||||
var walSegment string
|
||||
|
||||
if pm.config.UseWalForCheckpoint && pm.walManager != nil {
|
||||
if segWal, ok := pm.walManager.(*SegmentedWALManager); ok {
|
||||
if segWal.currentSegment != nil {
|
||||
segWal.currentSegment.Writer.Flush()
|
||||
RealFsync(segWal.currentSegment.File)
|
||||
walLSN = segWal.currentSegment.EndLSN
|
||||
walSegment = fmt.Sprintf("%d", segWal.currentSegment.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
snapshot := &DatabaseSnapshot{
|
||||
Name: dbName,
|
||||
Collections: make(map[string]interface{}),
|
||||
CreatedAt: time.Now().UnixMilli(),
|
||||
Version: pm.checkpointID.Add(1),
|
||||
WalLSN: walLSN,
|
||||
WalSegment: walSegment,
|
||||
}
|
||||
|
||||
for _, collName := range db.ListCollections() {
|
||||
@@ -170,19 +431,28 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error {
|
||||
"fields": doc.GetFields(),
|
||||
"created_at": doc.CreatedAt,
|
||||
"updated_at": doc.UpdatedAt,
|
||||
"deleted_at": doc.DeletedAt,
|
||||
"version": doc.Version,
|
||||
})
|
||||
}
|
||||
snapshot.Collections[collName] = collData
|
||||
}
|
||||
|
||||
// Сериализуем и сохраняем
|
||||
data, err := json.Marshal(snapshot)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Опционально сжимаем
|
||||
// Вычисляем контрольную сумму
|
||||
hash := sha256.Sum256(data)
|
||||
snapshot.Checksum = hex.EncodeToString(hash[:])
|
||||
|
||||
// Пере-сериализуем с контрольной суммой
|
||||
data, err = json.Marshal(snapshot)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if pm.config.CompressEnabled {
|
||||
data, err = pm.compress(data)
|
||||
if err != nil {
|
||||
@@ -190,29 +460,61 @@ func (pm *PersistenceManager) SaveDatabase(dbName string) error {
|
||||
}
|
||||
}
|
||||
|
||||
// Сохраняем в файл
|
||||
// Атомарная запись с использованием временного файла
|
||||
filename := pm.getSnapshotFilename(dbName, snapshot.Version)
|
||||
if err := os.WriteFile(filename, data, 0644); err != nil {
|
||||
if err := pm.atomicWriteFile(filename, data); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Синхронизируем с диском если нужно
|
||||
if pm.config.SyncWrites {
|
||||
if f, err := os.OpenFile(filename, os.O_RDWR, 0644); err == nil {
|
||||
RealFsync(f)
|
||||
f.Close()
|
||||
}
|
||||
}
|
||||
pm.lastCheckpoint = snapshot.CreatedAt
|
||||
|
||||
if pm.logger != nil {
|
||||
pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes)", dbName, snapshot.Version, len(data)))
|
||||
pm.logger.Info(fmt.Sprintf("Saved database %s checkpoint %d (%d bytes, checksum: %s, WAL LSN: %d)",
|
||||
dbName, snapshot.Version, len(data), snapshot.Checksum[:8], walLSN))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// atomicWriteFile выполняет атомарную запись файла
|
||||
func (pm *PersistenceManager) atomicWriteFile(filename string, data []byte) error {
|
||||
if pm.config.AtomicWrites {
|
||||
// Создаём временный файл в той же директории
|
||||
tempFile := filename + ".tmp"
|
||||
if err := os.WriteFile(tempFile, data, 0644); err != nil {
|
||||
return fmt.Errorf("failed to write temp file: %v", err)
|
||||
}
|
||||
|
||||
// Синхронизируем временный файл
|
||||
if pm.config.SyncWrites {
|
||||
if f, err := os.OpenFile(tempFile, os.O_RDWR, 0644); err == nil {
|
||||
RealFsync(f)
|
||||
f.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// Атомарное переименование
|
||||
if err := os.Rename(tempFile, filename); err != nil {
|
||||
os.Remove(tempFile)
|
||||
return fmt.Errorf("failed to rename temp file: %v", err)
|
||||
}
|
||||
|
||||
// Синхронизируем директорию
|
||||
FsyncDir(pm.config.DataDir)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Обычная запись (не атомарная)
|
||||
return os.WriteFile(filename, data, 0644)
|
||||
}
|
||||
|
||||
// SaveAll сохраняет все базы данных
|
||||
func (pm *PersistenceManager) SaveAll() error {
|
||||
if pm.isRestoring.Load() {
|
||||
return fmt.Errorf("cannot save checkpoints during restore")
|
||||
}
|
||||
|
||||
databases := pm.storage.ListDatabases()
|
||||
|
||||
var lastErr error
|
||||
@@ -228,44 +530,80 @@ func (pm *PersistenceManager) SaveAll() error {
|
||||
return lastErr
|
||||
}
|
||||
|
||||
// LoadDatabase загружает базу данных с диска
|
||||
// LoadDatabase загружает базу данных с диска с применением WAL
|
||||
func (pm *PersistenceManager) LoadDatabase(dbName string) error {
|
||||
pm.isRestoring.Store(true)
|
||||
defer pm.isRestoring.Store(false)
|
||||
|
||||
pm.mu.Lock()
|
||||
defer pm.mu.Unlock()
|
||||
|
||||
// Находим последний чекпоинт
|
||||
startTime := time.Now()
|
||||
|
||||
recoveryInfo := &CheckpointRecoveryInfo{
|
||||
CheckpointTime: time.Now().UnixMilli(),
|
||||
Success: false,
|
||||
}
|
||||
|
||||
snapshot, err := pm.findLatestSnapshot(dbName)
|
||||
if err != nil {
|
||||
recoveryInfo.Error = err.Error()
|
||||
pm.recoveryInfo = recoveryInfo
|
||||
return err
|
||||
}
|
||||
|
||||
if snapshot == nil {
|
||||
recoveryInfo.Error = "no snapshot found"
|
||||
pm.recoveryInfo = recoveryInfo
|
||||
return fmt.Errorf("no snapshot found for database: %s", dbName)
|
||||
}
|
||||
|
||||
// Восстанавливаем базу данных
|
||||
recoveryInfo.CheckpointVersion = snapshot.Version
|
||||
recoveryInfo.WalLSN = snapshot.WalLSN
|
||||
|
||||
// Проверяем целостность чекпоинта
|
||||
if err := pm.verifySnapshot(snapshot); err != nil {
|
||||
recoveryInfo.Error = err.Error()
|
||||
pm.recoveryInfo = recoveryInfo
|
||||
return fmt.Errorf("snapshot integrity check failed: %v", err)
|
||||
}
|
||||
|
||||
// Создаём базу данных если не существует
|
||||
if !pm.storage.ExistsDatabase(dbName) {
|
||||
if err := pm.storage.CreateDatabase(dbName); err != nil {
|
||||
recoveryInfo.Error = err.Error()
|
||||
pm.recoveryInfo = recoveryInfo
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
db, err := pm.storage.GetDatabase(dbName)
|
||||
if err != nil {
|
||||
recoveryInfo.Error = err.Error()
|
||||
pm.recoveryInfo = recoveryInfo
|
||||
return err
|
||||
}
|
||||
|
||||
// Восстанавливаем коллекции
|
||||
restoredColls := 0
|
||||
restoredDocs := int64(0)
|
||||
|
||||
for collName, collDataRaw := range snapshot.Collections {
|
||||
collData, ok := collDataRaw.([]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
|
||||
if _, err := db.GetCollection(collName); err != nil {
|
||||
if err := db.CreateCollection(collName); err != nil {
|
||||
continue
|
||||
// Удаляем существующую коллекцию если есть
|
||||
if _, err := db.GetCollection(collName); err == nil {
|
||||
db.DropCollection(collName)
|
||||
}
|
||||
|
||||
if err := db.CreateCollection(collName); err != nil {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Error(fmt.Sprintf("Failed to create collection %s: %v", collName, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
coll, err := db.GetCollection(collName)
|
||||
@@ -273,6 +611,8 @@ func (pm *PersistenceManager) LoadDatabase(dbName string) error {
|
||||
continue
|
||||
}
|
||||
|
||||
restoredColls++
|
||||
|
||||
for _, docRaw := range collData {
|
||||
docMap, ok := docRaw.(map[string]interface{})
|
||||
if !ok {
|
||||
@@ -294,26 +634,176 @@ func (pm *PersistenceManager) LoadDatabase(dbName string) error {
|
||||
|
||||
if createdAt, ok := docMap["created_at"].(float64); ok {
|
||||
doc.CreatedAt = int64(createdAt)
|
||||
} else if createdAt, ok := docMap["created_at"].(int64); ok {
|
||||
doc.CreatedAt = createdAt
|
||||
}
|
||||
|
||||
if updatedAt, ok := docMap["updated_at"].(float64); ok {
|
||||
doc.UpdatedAt = int64(updatedAt)
|
||||
} else if updatedAt, ok := docMap["updated_at"].(int64); ok {
|
||||
doc.UpdatedAt = updatedAt
|
||||
}
|
||||
|
||||
if deletedAt, ok := docMap["deleted_at"].(float64); ok {
|
||||
doc.DeletedAt = int64(deletedAt)
|
||||
} else if deletedAt, ok := docMap["deleted_at"].(int64); ok {
|
||||
doc.DeletedAt = deletedAt
|
||||
}
|
||||
|
||||
if version, ok := docMap["version"].(float64); ok {
|
||||
doc.Version = uint64(version)
|
||||
} else if version, ok := docMap["version"].(uint64); ok {
|
||||
doc.Version = version
|
||||
}
|
||||
|
||||
if err := coll.Insert(doc); err != nil {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Error(fmt.Sprintf("Failed to restore document %s: %v", docID, err))
|
||||
pm.logger.Warn(fmt.Sprintf("Failed to restore document %s: %v", docID, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
restoredDocs++
|
||||
}
|
||||
}
|
||||
|
||||
recoveryInfo.RestoredDocs = restoredDocs
|
||||
recoveryInfo.RestoredColls = restoredColls
|
||||
|
||||
// Применяем WAL записи после чекпоинта
|
||||
appliedWalCount := 0
|
||||
if pm.config.UseWalForCheckpoint && snapshot.WalLSN > 0 && pm.walReader != nil {
|
||||
currentLSN, err := pm.walReader.GetCurrentIndex()
|
||||
if err == nil && currentLSN > snapshot.WalLSN {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Info(fmt.Sprintf("Database %s: Applying %d WAL entries after checkpoint (LSN %d -> %d)",
|
||||
dbName, currentLSN-snapshot.WalLSN, snapshot.WalLSN, currentLSN))
|
||||
}
|
||||
|
||||
entries, err := pm.walReader.ReadSince(snapshot.WalLSN)
|
||||
if err != nil {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Error(fmt.Sprintf("Failed to read WAL entries: %v", err))
|
||||
}
|
||||
} else {
|
||||
for _, entry := range entries {
|
||||
if err := pm.applyWALEntry(db, entry); err != nil {
|
||||
if pm.logger != nil {
|
||||
pm.logger.Error(fmt.Sprintf("Failed to apply WAL entry %d: %v", entry.Index, err))
|
||||
}
|
||||
continue
|
||||
}
|
||||
appliedWalCount++
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
recoveryInfo.AppliedWalCount = appliedWalCount
|
||||
recoveryInfo.DurationMs = time.Since(startTime).Milliseconds()
|
||||
recoveryInfo.Success = true
|
||||
|
||||
pm.recoveryInfo = recoveryInfo
|
||||
|
||||
if pm.logger != nil {
|
||||
pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d)", dbName, snapshot.Version))
|
||||
pm.logger.Info(fmt.Sprintf("Loaded database %s from snapshot (version %d, WAL LSN: %d, applied %d WAL entries, %d docs, %d colls, duration: %dms)",
|
||||
dbName, snapshot.Version, snapshot.WalLSN, appliedWalCount, restoredDocs, restoredColls, recoveryInfo.DurationMs))
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// applyWALEntry применяет одну запись WAL к базе данных
|
||||
func (pm *PersistenceManager) applyWALEntry(db *Database, entry WALEntry) error {
|
||||
// Проверяем тип записи (1 = Transaction) - ИСПРАВЛЕНО
|
||||
if entry.Type != "1" {
|
||||
return nil
|
||||
}
|
||||
|
||||
var txRecord struct {
|
||||
ID uint64 `json:"id"`
|
||||
State int32 `json:"state"`
|
||||
Timestamp int64 `json:"timestamp"`
|
||||
Operations []struct {
|
||||
Type string `json:"type"`
|
||||
Database string `json:"database"`
|
||||
Collection string `json:"collection"`
|
||||
DocumentID string `json:"document_id"`
|
||||
Data map[string]interface{} `json:"data"`
|
||||
Version uint64 `json:"version"`
|
||||
} `json:"operations"`
|
||||
}
|
||||
|
||||
if err := json.Unmarshal(entry.Data, &txRecord); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Применяем только закоммиченные транзакции (state = 1)
|
||||
if txRecord.State != 1 { // TransactionCommitted
|
||||
return nil
|
||||
}
|
||||
|
||||
for _, op := range txRecord.Operations {
|
||||
if op.Database != db.name {
|
||||
continue
|
||||
}
|
||||
|
||||
coll, err := db.GetCollection(op.Collection)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
switch op.Type {
|
||||
case "insert":
|
||||
doc := NewDocumentWithID(op.DocumentID)
|
||||
for k, v := range op.Data {
|
||||
doc.SetField(k, v)
|
||||
}
|
||||
doc.Version = op.Version
|
||||
if err := coll.Insert(doc); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
case "update":
|
||||
if err := coll.Update(op.DocumentID, op.Data); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
case "delete":
|
||||
if err := coll.Delete(op.DocumentID); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// verifySnapshot проверяет целостность снапшота
|
||||
func (pm *PersistenceManager) verifySnapshot(snapshot *DatabaseSnapshot) error {
|
||||
if snapshot.Checksum == "" {
|
||||
return fmt.Errorf("snapshot has no checksum")
|
||||
}
|
||||
|
||||
// Пересоздаём данные без контрольной суммы для проверки
|
||||
tempSnapshot := &DatabaseSnapshot{
|
||||
Name: snapshot.Name,
|
||||
Collections: snapshot.Collections,
|
||||
CreatedAt: snapshot.CreatedAt,
|
||||
Version: snapshot.Version,
|
||||
WalLSN: snapshot.WalLSN,
|
||||
WalSegment: snapshot.WalSegment,
|
||||
}
|
||||
|
||||
data, err := json.Marshal(tempSnapshot)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal for checksum: %v", err)
|
||||
}
|
||||
|
||||
hash := sha256.Sum256(data)
|
||||
checksum := hex.EncodeToString(hash[:])
|
||||
|
||||
if checksum != snapshot.Checksum {
|
||||
return fmt.Errorf("checksum mismatch: expected %s, got %s", snapshot.Checksum[:8], checksum[:8])
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -329,8 +819,6 @@ func (pm *PersistenceManager) LoadAll() error {
|
||||
databases := make(map[string]bool)
|
||||
for _, file := range files {
|
||||
base := filepath.Base(file)
|
||||
// Извлекаем имя базы данных из имени файла
|
||||
// Формат: snapshot_{dbname}_{version}.json.gz
|
||||
parts := strings.Split(base, "_")
|
||||
if len(parts) >= 2 {
|
||||
dbName := parts[1]
|
||||
@@ -370,7 +858,6 @@ func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapsh
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Сортируем по времени создания (новейшие первыми)
|
||||
sort.Slice(files, func(i, j int) bool {
|
||||
infoI, _ := os.Stat(files[i])
|
||||
infoJ, _ := os.Stat(files[j])
|
||||
@@ -380,7 +867,6 @@ func (pm *PersistenceManager) findLatestSnapshot(dbName string) (*DatabaseSnapsh
|
||||
return infoI.ModTime().After(infoJ.ModTime())
|
||||
})
|
||||
|
||||
// Читаем последний снапшот
|
||||
data, err := os.ReadFile(files[0])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -445,7 +931,6 @@ func (pm *PersistenceManager) cleanupOldCheckpoints() {
|
||||
return
|
||||
}
|
||||
|
||||
// Сортируем по времени модификации
|
||||
sort.Slice(files, func(i, j int) bool {
|
||||
infoI, _ := os.Stat(files[i])
|
||||
infoJ, _ := os.Stat(files[j])
|
||||
@@ -455,7 +940,6 @@ func (pm *PersistenceManager) cleanupOldCheckpoints() {
|
||||
return infoI.ModTime().Before(infoJ.ModTime())
|
||||
})
|
||||
|
||||
// Удаляем старые
|
||||
toDelete := files[:len(files)-pm.config.MaxCheckpoints]
|
||||
for _, f := range toDelete {
|
||||
os.Remove(f)
|
||||
@@ -464,3 +948,37 @@ func (pm *PersistenceManager) cleanupOldCheckpoints() {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// GetLastCheckpointInfo возвращает информацию о последнем чекпоинте
|
||||
func (pm *PersistenceManager) GetLastCheckpointInfo() map[string]interface{} {
|
||||
pm.mu.RLock()
|
||||
defer pm.mu.RUnlock()
|
||||
|
||||
info := map[string]interface{}{
|
||||
"last_checkpoint_time": pm.lastCheckpoint,
|
||||
"last_checkpoint_time_str": time.UnixMilli(pm.lastCheckpoint).Format("2006-01-02 15:04:05.000"),
|
||||
"checkpoint_id": pm.checkpointID.Load(),
|
||||
"wal_path": pm.config.WalPath,
|
||||
"use_wal": pm.config.UseWalForCheckpoint,
|
||||
"atomic_writes": pm.config.AtomicWrites,
|
||||
"is_restoring": pm.isRestoring.Load(),
|
||||
}
|
||||
|
||||
if pm.recoveryInfo != nil {
|
||||
info["last_recovery"] = pm.recoveryInfo
|
||||
}
|
||||
|
||||
return info
|
||||
}
|
||||
|
||||
// GetRecoveryInfo возвращает информацию о последнем восстановлении
|
||||
func (pm *PersistenceManager) GetRecoveryInfo() *CheckpointRecoveryInfo {
|
||||
pm.mu.RLock()
|
||||
defer pm.mu.RUnlock()
|
||||
return pm.recoveryInfo
|
||||
}
|
||||
|
||||
// IsRestoring возвращает статус восстановления
|
||||
func (pm *PersistenceManager) IsRestoring() bool {
|
||||
return pm.isRestoring.Load()
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user