forked from dlang/cdcdb
Переписана под ООП
This commit is contained in:
parent
46138c032a
commit
8a9142234e
GPG key ID:
C8D8BE544A27C511
14 changed files with 372 additions and 526 deletions
3
dub.json
3
dub.json
|
|
@ -11,8 +11,7 @@
|
|||
"zstd": "~>0.2.1"
|
||||
},
|
||||
"stringImportPaths": [
|
||||
"source/cdcdb/db",
|
||||
"source/cdcdb/cdc"
|
||||
"source/cdcdb"
|
||||
],
|
||||
"configurations": [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -1,185 +0,0 @@
|
|||
module cdcdb.cdc.cas;
|
||||
|
||||
import cdcdb.db;
|
||||
import cdcdb.cdc.core;
|
||||
|
||||
import zstd;
|
||||
|
||||
import std.digest.sha : SHA256, digest;
|
||||
import std.format : format;
|
||||
import std.exception : enforce;
|
||||
|
||||
// Content-Addressable Storage (Контентно-адресуемая система хранения)
|
||||
// CAS-хранилище со снапшотами
|
||||
final class CAS
|
||||
{
|
||||
private:
|
||||
DBLite _db;
|
||||
bool _zstd;
|
||||
|
||||
size_t _minSize;
|
||||
size_t _normalSize;
|
||||
size_t _maxSize;
|
||||
size_t _maskS;
|
||||
size_t _maskL;
|
||||
CDC _cdc;
|
||||
public:
|
||||
this(
|
||||
string database,
|
||||
bool zstd = false,
|
||||
size_t busyTimeout = 3000,
|
||||
size_t maxRetries = 3,
|
||||
size_t minSize = 256,
|
||||
size_t normalSize = 512,
|
||||
size_t maxSize = 1024,
|
||||
size_t maskS = 0xFF,
|
||||
size_t maskL = 0x0F
|
||||
) {
|
||||
_db = new DBLite(database, busyTimeout, maxRetries);
|
||||
_zstd = zstd;
|
||||
|
||||
_minSize = minSize;
|
||||
_normalSize = normalSize;
|
||||
_maxSize = maxSize;
|
||||
_maskS = maskS;
|
||||
_maskL = maskL;
|
||||
|
||||
_cdc = new CDC(_minSize, _normalSize, _maxSize, _maskS, _maskL);
|
||||
}
|
||||
|
||||
size_t newSnapshot(string label, const(ubyte)[] data, string description = string.init)
|
||||
{
|
||||
if (data.length == 0) {
|
||||
throw new Exception("Данные имеют нулевой размер");
|
||||
}
|
||||
|
||||
ubyte[32] sha256 = digest!SHA256(data);
|
||||
|
||||
// Если последний снимок файла соответствует текущему состоянию
|
||||
if (_db.isLast(label, sha256)) return 0;
|
||||
|
||||
Snapshot snapshot;
|
||||
|
||||
snapshot.label = label;
|
||||
snapshot.sha256 = sha256;
|
||||
snapshot.description = description;
|
||||
snapshot.sourceLength = data.length;
|
||||
snapshot.algoMin = _minSize;
|
||||
snapshot.algoNormal = _normalSize;
|
||||
snapshot.algoMax = _maxSize;
|
||||
snapshot.maskS = _maskS;
|
||||
snapshot.maskL = _maskL;
|
||||
|
||||
_db.beginImmediate();
|
||||
|
||||
bool ok;
|
||||
|
||||
scope (exit)
|
||||
{
|
||||
if (!ok)
|
||||
_db.rollback();
|
||||
}
|
||||
scope (success)
|
||||
{
|
||||
_db.commit();
|
||||
}
|
||||
|
||||
auto idSnapshot = _db.addSnapshot(snapshot);
|
||||
|
||||
SnapshotChunk snapshotChunk;
|
||||
Blob blob;
|
||||
|
||||
blob.zstd = _zstd;
|
||||
|
||||
// Разбить на фрагменты
|
||||
Chunk[] chunks = _cdc.split(data);
|
||||
|
||||
// Запись фрагментов в БД
|
||||
foreach (chunk; chunks)
|
||||
{
|
||||
blob.sha256 = chunk.sha256;
|
||||
blob.size = chunk.size;
|
||||
|
||||
auto content = data[chunk.offset .. chunk.offset + chunk.size];
|
||||
|
||||
if (_zstd) {
|
||||
ubyte[] zBytes = compress(content, 22);
|
||||
size_t zSize = zBytes.length;
|
||||
ubyte[32] zHash = digest!SHA256(zBytes);
|
||||
|
||||
blob.zSize = zSize;
|
||||
blob.zSha256 = zHash;
|
||||
blob.content = zBytes;
|
||||
} else {
|
||||
blob.content = content.dup;
|
||||
}
|
||||
|
||||
// Запись фрагментов
|
||||
_db.addBlob(blob);
|
||||
|
||||
snapshotChunk.snapshotId = idSnapshot;
|
||||
snapshotChunk.chunkIndex = chunk.index;
|
||||
snapshotChunk.offset = chunk.offset;
|
||||
snapshotChunk.sha256 = chunk.sha256;
|
||||
|
||||
// Привязка фрагментов к снимку
|
||||
_db.addSnapshotChunk(snapshotChunk);
|
||||
}
|
||||
|
||||
ok = true;
|
||||
|
||||
return idSnapshot;
|
||||
}
|
||||
|
||||
Snapshot[] getSnapshots(string label = string.init)
|
||||
{
|
||||
return _db.getSnapshots(label);
|
||||
}
|
||||
|
||||
ubyte[] getSnapshotData(const ref Snapshot snapshot)
|
||||
{
|
||||
auto dataChunks = _db.getChunks(snapshot.id);
|
||||
ubyte[] content;
|
||||
|
||||
foreach (chunk; dataChunks) {
|
||||
ubyte[] bytes;
|
||||
if (chunk.zstd) {
|
||||
enforce(chunk.zSize == chunk.content.length, "Размер сжатого фрагмента не соответствует ожидаемому");
|
||||
bytes = cast(ubyte[]) uncompress(chunk.content);
|
||||
} else {
|
||||
bytes = chunk.content;
|
||||
}
|
||||
enforce(chunk.size == bytes.length, "Оригинальный размер не соответствует ожидаемому");
|
||||
content ~= bytes;
|
||||
}
|
||||
enforce(snapshot.sha256 == digest!SHA256(content), "Хеш-сумма файла не совпадает");
|
||||
|
||||
return content;
|
||||
}
|
||||
|
||||
void removeSnapshot(const ref Snapshot snapshot)
|
||||
{
|
||||
_db.beginImmediate();
|
||||
|
||||
bool ok;
|
||||
|
||||
scope (exit)
|
||||
{
|
||||
if (!ok)
|
||||
_db.rollback();
|
||||
}
|
||||
scope (success)
|
||||
{
|
||||
_db.commit();
|
||||
}
|
||||
|
||||
_db.deleteSnapshot(snapshot.id);
|
||||
|
||||
ok = true;
|
||||
}
|
||||
|
||||
string getVersion() const @safe nothrow {
|
||||
import cdcdb.version_;
|
||||
return cdcdbVersion;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
module cdcdb.cdc;
|
||||
|
||||
public import cdcdb.cdc.cas;
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
module cdcdb.cdc.core;
|
||||
module cdcdb.core;
|
||||
|
||||
import std.digest.sha : SHA256, digest;
|
||||
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
module cdcdb.db;
|
||||
|
||||
public import cdcdb.db.dblite;
|
||||
public import cdcdb.db.types;
|
||||
|
|
@ -1,197 +0,0 @@
|
|||
# Схемы базы данных для хранения снимков (фрагментов)
|
||||
|
||||
## Структура базы данных
|
||||
```mermaid
|
||||
erDiagram
|
||||
%% Композитный PK у SNAPSHOT_CHUNKS: (snapshot_id, chunk_index)
|
||||
|
||||
SNAPSHOTS {
|
||||
int id PK
|
||||
string label
|
||||
string created_utc
|
||||
int source_length
|
||||
int algo_min
|
||||
int algo_normal
|
||||
int algo_max
|
||||
int mask_s
|
||||
int mask_l
|
||||
string status
|
||||
}
|
||||
|
||||
BLOBS {
|
||||
string sha256 PK
|
||||
int size
|
||||
blob content
|
||||
string created_utc
|
||||
}
|
||||
|
||||
SNAPSHOT_CHUNKS {
|
||||
int snapshot_id FK
|
||||
int chunk_index
|
||||
int offset
|
||||
int size
|
||||
string sha256 FK
|
||||
}
|
||||
|
||||
%% Связи и поведение внешних ключей
|
||||
SNAPSHOTS ||--o{ SNAPSHOT_CHUNKS : "1:N, ON DELETE CASCADE"
|
||||
BLOBS ||--o{ SNAPSHOT_CHUNKS : "1:N, ON DELETE RESTRICT"
|
||||
```
|
||||
|
||||
## Схема последовательности записи в базу данных
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
autonumber
|
||||
participant APP as Приложение
|
||||
participant CH as Разбиение на чанки (FastCDC)
|
||||
participant HS as Хеширование (SHA-256)
|
||||
participant DB as База данных (SQLite)
|
||||
|
||||
Note over APP,DB: Подготовка
|
||||
APP->>DB: Открывает соединение, включает PRAGMA (WAL, foreign_keys=ON)
|
||||
APP->>DB: BEGIN IMMEDIATE (начать транзакцию с блокировкой на запись)
|
||||
|
||||
Note over APP,DB: Создание метаданных снимка
|
||||
APP->>DB: INSERT INTO snapshots(label, source_length, algo_min, algo_normal, algo_max, mask_s, mask_l, status='pending')
|
||||
DB-->>APP: id снимка = last_insert_rowid()
|
||||
|
||||
Note over APP,CH: Поток файла → чанки
|
||||
APP->>CH: Читает файл, передает параметры FastCDC (min/normal/max, mask_s/mask_l)
|
||||
loop Для каждого чанка в порядке следования
|
||||
CH-->>APP: Возвращает {chunk_index, offset, size, bytes}
|
||||
|
||||
Note over APP,HS: Хеш содержимого
|
||||
APP->>HS: Вычисляет SHA-256(bytes)
|
||||
HS-->>APP: digest (sha256)
|
||||
|
||||
Note over APP,DB: Дедупликация контента
|
||||
APP->>DB: SELECT 1 FROM blobs WHERE sha256 = ?
|
||||
alt Блоб отсутствует
|
||||
APP->>DB: INSERT INTO blobs(sha256, size, content)
|
||||
DB-->>APP: OK
|
||||
else Блоб уже есть
|
||||
DB-->>APP: Найден (пропускаем вставку содержимого)
|
||||
end
|
||||
|
||||
Note over APP,DB: Привязка чанка к снимку
|
||||
APP->>DB: INSERT INTO snapshot_chunks(snapshot_id, chunk_index, offset, size, sha256)
|
||||
DB-->>APP: OK (PK: (snapshot_id, chunk_index))
|
||||
end
|
||||
|
||||
Note over APP,DB: Валидация и завершение
|
||||
APP->>DB: SELECT SUM(size) FROM snapshot_chunks WHERE snapshot_id = ?
|
||||
DB-->>APP: total_size
|
||||
alt total_size == snapshots.source_length
|
||||
APP->>DB: UPDATE snapshots SET status='ready' WHERE id = ?
|
||||
APP->>DB: COMMIT
|
||||
DB-->>APP: Транзакция зафиксирована
|
||||
else Несоответствие размеров или ошибка
|
||||
APP->>DB: ROLLBACK
|
||||
DB-->>APP: Откат изменений
|
||||
APP-->>APP: Логирует ошибку, возвращает код/исключение
|
||||
end
|
||||
```
|
||||
|
||||
## Схема последовательности восстановления из базы данных
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
autonumber
|
||||
participant APP as Приложение
|
||||
participant DB as База данных (SQLite)
|
||||
participant FS as Целевой файл
|
||||
participant HS as Хеширование (опц.)
|
||||
|
||||
Note over APP,DB: Подготовка к чтению
|
||||
APP->>DB: Открывает соединение (read), BEGIN (снимок чтения)
|
||||
|
||||
Note over APP,DB: Выбор снимка
|
||||
APP->>DB: Находит нужный снимок по id/label, читает status и source_length
|
||||
DB-->>APP: id, status, source_length
|
||||
alt status == "ready"
|
||||
else снимок не готов
|
||||
APP-->>APP: Прерывает восстановление с ошибкой
|
||||
DB-->>APP: END
|
||||
end
|
||||
|
||||
Note over APP,DB: Получение состава снимка
|
||||
APP->>DB: SELECT chunk_index, offset, size, sha256 FROM snapshot_chunks WHERE snapshot_id=? ORDER BY chunk_index
|
||||
DB-->>APP: Строки чанков в порядке chunk_index
|
||||
|
||||
loop Для каждого чанка
|
||||
APP->>DB: SELECT content, size FROM blobs WHERE sha256=?
|
||||
DB-->>APP: content, blob_size
|
||||
|
||||
Note over APP,HS: (опц.) контроль целостности чанка
|
||||
APP->>HS: Вычисляет SHA-256(content)
|
||||
HS-->>APP: digest
|
||||
APP-->>APP: Сверяет digest с sha256 и size с blob_size
|
||||
|
||||
alt offset задан
|
||||
APP->>FS: Позиционируется на offset и пишет content (pwrite/seek+write)
|
||||
else offset отсутствует
|
||||
APP->>FS: Дописывает content в конец файла
|
||||
end
|
||||
end
|
||||
|
||||
Note over APP,DB: Финальная проверка
|
||||
APP-->>APP: Суммирует размеры записанных чанков → total_size
|
||||
APP->>DB: Берёт snapshots.source_length
|
||||
DB-->>APP: source_length
|
||||
alt total_size == source_length
|
||||
APP->>FS: fsync и close
|
||||
DB-->>APP: END
|
||||
APP-->>APP: Успешное восстановление
|
||||
else размеры не совпали
|
||||
APP->>FS: Удаляет/помечает файл как повреждённый
|
||||
DB-->>APP: END
|
||||
APP-->>APP: Фиксирует ошибку (несоответствие сумм)
|
||||
end
|
||||
```
|
||||
|
||||
## Схема записи в БД
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
autonumber
|
||||
participant APP as Приложение
|
||||
participant DB as SQLite
|
||||
participant CH as Разбиение (FastCDC)
|
||||
participant HS as SHA-256
|
||||
|
||||
Note over APP,DB: Подготовка к записи
|
||||
APP->>DB: PRAGMA foreign_keys=ON
|
||||
APP->>DB: BEGIN IMMEDIATE
|
||||
|
||||
Note over APP,DB: Метаданные снимка
|
||||
APP->>DB: INSERT INTO snapshots(..., status='pending')
|
||||
DB-->>APP: snap_id := last_insert_rowid()
|
||||
|
||||
Note over APP,CH: Поток файла → чанки (min/normal/max, mask_s/mask_l)
|
||||
loop Для каждого чанка по порядку
|
||||
CH-->>APP: {chunk_index, offset, size, bytes}
|
||||
|
||||
Note over APP,HS: Хеширование
|
||||
APP->>HS: SHA-256(bytes)
|
||||
HS-->>APP: sha256 (32 байта)
|
||||
|
||||
Note over APP,DB: Дедупликация содержимого
|
||||
APP->>DB: INSERT INTO blobs(sha256,size,content) ON CONFLICT DO NOTHING
|
||||
DB-->>APP: OK (новая строка или уже была)
|
||||
|
||||
Note over APP,DB: Привязка к снимку
|
||||
APP->>DB: INSERT INTO snapshot_chunks(snapshot_id,chunk_index,offset,size,sha256)
|
||||
DB-->>APP: OK (триггер ++refcount, last_seen_utc=now)
|
||||
end
|
||||
|
||||
Note over APP,DB: Валидация и финал
|
||||
APP->>DB: SELECT SUM(size) FROM snapshot_chunks WHERE snapshot_id = snap_id
|
||||
DB-->>APP: total_size
|
||||
alt total_size == snapshots.source_length
|
||||
Note over DB: триггер mark_ready ставит status='ready'
|
||||
APP->>DB: COMMIT
|
||||
else несовпадение / ошибка
|
||||
APP->>DB: ROLLBACK
|
||||
end
|
||||
```
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
module cdcdb.db.types;
|
||||
|
||||
import std.datetime : DateTime;
|
||||
|
||||
enum SnapshotStatus : int
|
||||
{
|
||||
pending = 0,
|
||||
ready = 1
|
||||
}
|
||||
|
||||
struct Snapshot
|
||||
{
|
||||
long id;
|
||||
string label;
|
||||
ubyte[32] sha256;
|
||||
string description;
|
||||
DateTime createdUtc;
|
||||
long sourceLength;
|
||||
long algoMin;
|
||||
long algoNormal;
|
||||
long algoMax;
|
||||
long maskS;
|
||||
long maskL;
|
||||
SnapshotStatus status;
|
||||
}
|
||||
|
||||
struct Blob
|
||||
{
|
||||
ubyte[32] sha256;
|
||||
ubyte[32] zSha256;
|
||||
long size;
|
||||
long zSize;
|
||||
ubyte[] content;
|
||||
DateTime createdUtc;
|
||||
DateTime lastSeenUtc;
|
||||
long refcount;
|
||||
bool zstd;
|
||||
}
|
||||
|
||||
struct SnapshotChunk
|
||||
{
|
||||
long snapshotId;
|
||||
long chunkIndex;
|
||||
long offset;
|
||||
ubyte[32] sha256;
|
||||
}
|
||||
|
||||
struct SnapshotDataChunk {
|
||||
long chunkIndex;
|
||||
long offset;
|
||||
long size;
|
||||
ubyte[] content;
|
||||
bool zstd;
|
||||
long zSize;
|
||||
ubyte[32] sha256;
|
||||
ubyte[32] zSha256;
|
||||
}
|
||||
|
|
@ -1,14 +1,66 @@
|
|||
module cdcdb.db.dblite;
|
||||
module cdcdb.dblite;
|
||||
|
||||
import cdcdb.db.types;
|
||||
import arsd.sqlite : Sqlite, SqliteResult, DatabaseException;
|
||||
|
||||
import arsd.sqlite;
|
||||
|
||||
import std.exception : enforce;
|
||||
import std.conv : to;
|
||||
import std.datetime : DateTime;
|
||||
import std.string : join, replace, toLower;
|
||||
import std.algorithm : canFind;
|
||||
import std.conv : to;
|
||||
import std.format : format;
|
||||
import std.exception : enforce;
|
||||
|
||||
enum SnapshotStatus : ubyte
|
||||
{
|
||||
pending = 0,
|
||||
ready = 1
|
||||
}
|
||||
|
||||
struct DBSnapshot {
|
||||
long id;
|
||||
string label;
|
||||
ubyte[32] sha256;
|
||||
string description;
|
||||
DateTime createdUtc;
|
||||
long sourceLength;
|
||||
long algoMin;
|
||||
long algoNormal;
|
||||
long algoMax;
|
||||
long maskS;
|
||||
long maskL;
|
||||
SnapshotStatus status;
|
||||
}
|
||||
|
||||
struct DBSnapshotChunk
|
||||
{
|
||||
long snapshotId;
|
||||
long chunkIndex;
|
||||
long offset;
|
||||
ubyte[32] sha256;
|
||||
}
|
||||
|
||||
struct DBBlob
|
||||
{
|
||||
ubyte[32] sha256;
|
||||
ubyte[32] zSha256;
|
||||
long size;
|
||||
long zSize;
|
||||
ubyte[] content;
|
||||
DateTime createdUtc;
|
||||
DateTime lastSeenUtc;
|
||||
long refcount;
|
||||
bool zstd;
|
||||
}
|
||||
|
||||
struct DBSnapshotChunkData {
|
||||
long chunkIndex;
|
||||
long offset;
|
||||
long size;
|
||||
ubyte[] content;
|
||||
bool zstd;
|
||||
long zSize;
|
||||
ubyte[32] sha256;
|
||||
ubyte[32] zSha256;
|
||||
}
|
||||
|
||||
final class DBLite : Sqlite
|
||||
{
|
||||
|
|
@ -120,7 +172,25 @@ public:
|
|||
sql("ROLLBACK");
|
||||
}
|
||||
|
||||
long addSnapshot(Snapshot snapshot)
|
||||
bool isLast(string label, ubyte[] sha256) {
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
SELECT COALESCE(
|
||||
(SELECT (label = ? AND sha256 = ?)
|
||||
FROM snapshots
|
||||
ORDER BY created_utc DESC
|
||||
LIMIT 1),
|
||||
0
|
||||
) AS is_last;
|
||||
}, label, sha256
|
||||
);
|
||||
|
||||
if (!queryResult.empty())
|
||||
return queryResult.front()["is_last"].to!long > 0;
|
||||
return false;
|
||||
}
|
||||
|
||||
long addSnapshot(DBSnapshot snapshot)
|
||||
{
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
|
|
@ -157,13 +227,14 @@ public:
|
|||
return queryResult.front()["id"].to!long;
|
||||
}
|
||||
|
||||
void addBlob(Blob blob)
|
||||
bool addBlob(DBBlob blob)
|
||||
{
|
||||
sql(
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
INSERT INTO blobs (sha256, z_sha256, size, z_size, content, zstd)
|
||||
VALUES (?,?,?,?,?,?)
|
||||
ON CONFLICT (sha256) DO NOTHING
|
||||
RETURNING sha256
|
||||
},
|
||||
blob.sha256[],
|
||||
blob.zstd ? blob.zSha256[] : null,
|
||||
|
|
@ -172,76 +243,28 @@ public:
|
|||
blob.content,
|
||||
blob.zstd.to!int
|
||||
);
|
||||
|
||||
return !queryResult.empty();
|
||||
}
|
||||
|
||||
void addSnapshotChunk(SnapshotChunk snapshotChunk)
|
||||
bool addSnapshotChunk(DBSnapshotChunk snapshotChunk)
|
||||
{
|
||||
sql(
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
INSERT INTO snapshot_chunks (snapshot_id, chunk_index, offset, sha256)
|
||||
VALUES(?,?,?,?)
|
||||
RETURNING snapshot_id
|
||||
},
|
||||
snapshotChunk.snapshotId,
|
||||
snapshotChunk.chunkIndex,
|
||||
snapshotChunk.offset,
|
||||
snapshotChunk.sha256[]
|
||||
);
|
||||
|
||||
return !queryResult.empty();
|
||||
}
|
||||
|
||||
bool isLast(string label, ubyte[] sha256) {
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
SELECT COALESCE(
|
||||
(SELECT (label = ? AND sha256 = ?)
|
||||
FROM snapshots
|
||||
ORDER BY created_utc DESC
|
||||
LIMIT 1),
|
||||
0
|
||||
) AS is_last;
|
||||
}, label, sha256
|
||||
);
|
||||
|
||||
if (!queryResult.empty())
|
||||
return queryResult.front()["is_last"].to!long > 0;
|
||||
return false;
|
||||
}
|
||||
|
||||
Snapshot[] getSnapshots(string label)
|
||||
{
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
SELECT id, label, sha256, description, created_utc, source_length,
|
||||
algo_min, algo_normal, algo_max, mask_s, mask_l, status
|
||||
FROM snapshots WHERE (length(?) = 0 OR label = ?1);
|
||||
}, label
|
||||
);
|
||||
|
||||
Snapshot[] snapshots;
|
||||
|
||||
foreach (row; queryResult)
|
||||
{
|
||||
Snapshot snapshot;
|
||||
|
||||
snapshot.id = row["id"].to!long;
|
||||
snapshot.label = row["label"].to!string;
|
||||
snapshot.sha256 = cast(ubyte[]) row["sha256"].dup;
|
||||
snapshot.description = row["description"].to!string;
|
||||
snapshot.createdUtc = toDateTime(row["created_utc"].to!string);
|
||||
snapshot.sourceLength = row["source_length"].to!long;
|
||||
snapshot.algoMin = row["algo_min"].to!long;
|
||||
snapshot.algoNormal = row["algo_normal"].to!long;
|
||||
snapshot.algoMax = row["algo_max"].to!long;
|
||||
snapshot.maskS = row["mask_s"].to!long;
|
||||
snapshot.maskL = row["mask_l"].to!long;
|
||||
snapshot.status = cast(SnapshotStatus) row["status"].to!int;
|
||||
|
||||
snapshots ~= snapshot;
|
||||
}
|
||||
|
||||
return snapshots;
|
||||
}
|
||||
|
||||
Snapshot getSnapshot(long id)
|
||||
DBSnapshot getSnapshot(long id)
|
||||
{
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
|
|
@ -251,7 +274,7 @@ public:
|
|||
}, id
|
||||
);
|
||||
|
||||
Snapshot snapshot;
|
||||
DBSnapshot snapshot;
|
||||
|
||||
if (!queryResult.empty())
|
||||
{
|
||||
|
|
@ -274,11 +297,42 @@ public:
|
|||
return snapshot;
|
||||
}
|
||||
|
||||
void deleteSnapshot(long id) {
|
||||
sql("DELETE FROM snapshots WHERE id = ?", id);
|
||||
DBSnapshot[] getSnapshots(string label)
|
||||
{
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
SELECT id, label, sha256, description, created_utc, source_length,
|
||||
algo_min, algo_normal, algo_max, mask_s, mask_l, status
|
||||
FROM snapshots WHERE (length(?) = 0 OR label = ?1);
|
||||
}, label
|
||||
);
|
||||
|
||||
DBSnapshot[] snapshots;
|
||||
|
||||
foreach (row; queryResult)
|
||||
{
|
||||
DBSnapshot snapshot;
|
||||
|
||||
snapshot.id = row["id"].to!long;
|
||||
snapshot.label = row["label"].to!string;
|
||||
snapshot.sha256 = cast(ubyte[]) row["sha256"].dup;
|
||||
snapshot.description = row["description"].to!string;
|
||||
snapshot.createdUtc = toDateTime(row["created_utc"].to!string);
|
||||
snapshot.sourceLength = row["source_length"].to!long;
|
||||
snapshot.algoMin = row["algo_min"].to!long;
|
||||
snapshot.algoNormal = row["algo_normal"].to!long;
|
||||
snapshot.algoMax = row["algo_max"].to!long;
|
||||
snapshot.maskS = row["mask_s"].to!long;
|
||||
snapshot.maskL = row["mask_l"].to!long;
|
||||
snapshot.status = cast(SnapshotStatus) row["status"].to!int;
|
||||
|
||||
snapshots ~= snapshot;
|
||||
}
|
||||
|
||||
return snapshots;
|
||||
}
|
||||
|
||||
SnapshotDataChunk[] getChunks(long snapshotId)
|
||||
DBSnapshotChunkData[] getChunks(long snapshotId)
|
||||
{
|
||||
auto queryResult = sql(
|
||||
q{
|
||||
|
|
@ -291,11 +345,11 @@ public:
|
|||
}, snapshotId
|
||||
);
|
||||
|
||||
SnapshotDataChunk[] sdchs;
|
||||
DBSnapshotChunkData[] sdchs;
|
||||
|
||||
foreach (row; queryResult)
|
||||
{
|
||||
SnapshotDataChunk sdch;
|
||||
DBSnapshotChunkData sdch;
|
||||
|
||||
sdch.chunkIndex = row["chunk_index"].to!long;
|
||||
sdch.offset = row["offset"].to!long;
|
||||
|
|
@ -311,4 +365,14 @@ public:
|
|||
|
||||
return sdchs;
|
||||
}
|
||||
|
||||
long deleteSnapshot(long id) {
|
||||
auto queryResult = sql("DELETE FROM snapshots WHERE id = ? RETURNING id", id);
|
||||
|
||||
if (queryResult.empty()) {
|
||||
throw new Exception("Ошибка при удалении снимка из базы данных");
|
||||
}
|
||||
|
||||
return queryResult.front()["id"].to!long;
|
||||
}
|
||||
}
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
module cdcdb;
|
||||
|
||||
public import cdcdb.cdc;
|
||||
public import cdcdb.storage;
|
||||
public import cdcdb.snapshot;
|
||||
|
|
|
|||
68
source/cdcdb/snapshot.d
Normal file
68
source/cdcdb/snapshot.d
Normal file
|
|
@ -0,0 +1,68 @@
|
|||
module cdcdb.snapshot;
|
||||
|
||||
import cdcdb.dblite;
|
||||
|
||||
import std.exception : enforce;
|
||||
|
||||
final class Snapshot {
|
||||
private:
|
||||
DBLite _db;
|
||||
DBSnapshot _snapshot;
|
||||
public:
|
||||
this(DBLite dblite, DBSnapshot dbSnapshot) {
|
||||
_db = dblite;
|
||||
_snapshot = dbSnapshot;
|
||||
}
|
||||
|
||||
this(DBLite dblite, long idSnapshot) {
|
||||
_db = dblite;
|
||||
_snapshot = _db.getSnapshot(idSnapshot);
|
||||
}
|
||||
|
||||
ubyte[] data() {
|
||||
auto dataChunks = _db.getChunks(_snapshot.id);
|
||||
ubyte[] content;
|
||||
|
||||
import zstd : uncompress;
|
||||
|
||||
foreach (chunk; dataChunks) {
|
||||
ubyte[] bytes;
|
||||
if (chunk.zstd) {
|
||||
enforce(chunk.zSize == chunk.content.length, "Размер сжатого фрагмента не соответствует ожидаемому");
|
||||
bytes = cast(ubyte[]) uncompress(chunk.content);
|
||||
} else {
|
||||
bytes = chunk.content;
|
||||
}
|
||||
enforce(chunk.size == bytes.length, "Оригинальный размер не соответствует ожидаемому");
|
||||
content ~= bytes;
|
||||
}
|
||||
|
||||
import std.digest.sha : SHA256, digest;
|
||||
|
||||
enforce(_snapshot.sha256 == digest!SHA256(content), "Хеш-сумма файла не совпадает");
|
||||
|
||||
return content;
|
||||
}
|
||||
|
||||
bool remove() {
|
||||
_db.beginImmediate();
|
||||
|
||||
bool ok;
|
||||
|
||||
scope (exit)
|
||||
{
|
||||
if (!ok)
|
||||
_db.rollback();
|
||||
}
|
||||
scope (success)
|
||||
{
|
||||
_db.commit();
|
||||
}
|
||||
|
||||
long idDeleted = _db.deleteSnapshot(_snapshot.id);
|
||||
|
||||
ok = true;
|
||||
|
||||
return _snapshot.id == idDeleted;
|
||||
}
|
||||
}
|
||||
154
source/cdcdb/storage.d
Normal file
154
source/cdcdb/storage.d
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
module cdcdb.storage;
|
||||
|
||||
import cdcdb.dblite;
|
||||
import cdcdb.core;
|
||||
import cdcdb.snapshot;
|
||||
|
||||
final class Storage
|
||||
{
|
||||
private:
|
||||
// Параметры работы с базой данных
|
||||
DBLite _db;
|
||||
bool _zstd;
|
||||
// Настройки CDC механизма
|
||||
CDC _cdc;
|
||||
size_t _minSize;
|
||||
size_t _normalSize;
|
||||
size_t _maxSize;
|
||||
size_t _maskS;
|
||||
size_t _maskL;
|
||||
|
||||
void initCDC(size_t minSize = 256, size_t normalSize = 512, size_t maxSize = 1024,
|
||||
size_t maskS = 0xFF, size_t maskL = 0x0F)
|
||||
{
|
||||
_minSize = minSize;
|
||||
_normalSize = normalSize;
|
||||
_maxSize = maxSize;
|
||||
_maskS = maskS;
|
||||
_maskL = maskL;
|
||||
// CDC не хранит динамически выделенных данных, переинициализация безопасна
|
||||
_cdc = new CDC(_minSize, _normalSize, _maxSize, _maskS, _maskL);
|
||||
}
|
||||
|
||||
public:
|
||||
this(string database, bool zstd = false, size_t busyTimeout = 3000, size_t maxRetries = 3)
|
||||
{
|
||||
_db = new DBLite(database, busyTimeout, maxRetries);
|
||||
_zstd = zstd;
|
||||
initCDC();
|
||||
}
|
||||
|
||||
void setupCDC(size_t minSize, size_t normalSize, size_t maxSize, size_t maskS, size_t maskL)
|
||||
{
|
||||
initCDC(minSize, normalSize, maxSize, maskS, maskL);
|
||||
}
|
||||
|
||||
Snapshot newSnapshot(string label, const(ubyte)[] data, string description = string.init)
|
||||
{
|
||||
if (data.length == 0)
|
||||
{
|
||||
throw new Exception("Данные имеют нулевой размер");
|
||||
}
|
||||
|
||||
import std.digest.sha : SHA256, digest;
|
||||
|
||||
ubyte[32] sha256 = digest!SHA256(data);
|
||||
|
||||
// Если последний снимок файла соответствует текущему состоянию
|
||||
if (_db.isLast(label, sha256))
|
||||
return null;
|
||||
|
||||
DBSnapshot dbSnapshot;
|
||||
|
||||
dbSnapshot.label = label;
|
||||
dbSnapshot.sha256 = sha256;
|
||||
dbSnapshot.description = description;
|
||||
dbSnapshot.sourceLength = data.length;
|
||||
dbSnapshot.algoMin = _minSize;
|
||||
dbSnapshot.algoNormal = _normalSize;
|
||||
dbSnapshot.algoMax = _maxSize;
|
||||
dbSnapshot.maskS = _maskS;
|
||||
dbSnapshot.maskL = _maskL;
|
||||
|
||||
_db.beginImmediate();
|
||||
|
||||
bool ok;
|
||||
|
||||
scope (exit)
|
||||
{
|
||||
if (!ok)
|
||||
_db.rollback();
|
||||
}
|
||||
scope (success)
|
||||
{
|
||||
_db.commit();
|
||||
}
|
||||
|
||||
auto idSnapshot = _db.addSnapshot(dbSnapshot);
|
||||
|
||||
DBSnapshotChunk dbSnapshotChunk;
|
||||
DBBlob dbBlob;
|
||||
|
||||
dbBlob.zstd = _zstd;
|
||||
|
||||
// Разбить на фрагменты
|
||||
Chunk[] chunks = _cdc.split(data);
|
||||
|
||||
import zstd : compress;
|
||||
|
||||
// Запись фрагментов в БД
|
||||
foreach (chunk; chunks)
|
||||
{
|
||||
dbBlob.sha256 = chunk.sha256;
|
||||
dbBlob.size = chunk.size;
|
||||
|
||||
auto content = data[chunk.offset .. chunk.offset + chunk.size];
|
||||
|
||||
if (_zstd) {
|
||||
ubyte[] zBytes = compress(content, 22);
|
||||
size_t zSize = zBytes.length;
|
||||
ubyte[32] zHash = digest!SHA256(zBytes);
|
||||
|
||||
dbBlob.zSize = zSize;
|
||||
dbBlob.zSha256 = zHash;
|
||||
dbBlob.content = zBytes;
|
||||
} else {
|
||||
dbBlob.content = content.dup;
|
||||
}
|
||||
|
||||
// Запись фрагментов
|
||||
_db.addBlob(dbBlob);
|
||||
|
||||
dbSnapshotChunk.snapshotId = idSnapshot;
|
||||
dbSnapshotChunk.chunkIndex = chunk.index;
|
||||
dbSnapshotChunk.offset = chunk.offset;
|
||||
dbSnapshotChunk.sha256 = chunk.sha256;
|
||||
|
||||
// Привязка фрагментов к снимку
|
||||
_db.addSnapshotChunk(dbSnapshotChunk);
|
||||
}
|
||||
|
||||
ok = true;
|
||||
|
||||
Snapshot snapshot = new Snapshot(_db, idSnapshot);
|
||||
|
||||
return snapshot;
|
||||
}
|
||||
|
||||
Snapshot[] getSnapshots(string label = string.init) {
|
||||
Snapshot[] snapshots;
|
||||
|
||||
foreach (snapshot; _db.getSnapshots(label)) {
|
||||
snapshots ~= new Snapshot(_db, snapshot);
|
||||
}
|
||||
|
||||
return snapshots;
|
||||
}
|
||||
|
||||
string getVersion() const @safe nothrow
|
||||
{
|
||||
import cdcdb.version_ : cdcdbVersion;
|
||||
|
||||
return cdcdbVersion;
|
||||
}
|
||||
}
|
||||
16
test/app.d
16
test/app.d
|
|
@ -6,12 +6,18 @@ import std.file : read;
|
|||
|
||||
void main()
|
||||
{
|
||||
auto cas = new CAS("/tmp/base.db", true);
|
||||
cas.newSnapshot("/tmp/text", cast(ubyte[]) read("/tmp/text"));
|
||||
// import std.stdio : writeln;
|
||||
auto storage = new Storage("/tmp/base.db", true);
|
||||
storage.newSnapshot("/tmp/text", cast(ubyte[]) read("/tmp/text"));
|
||||
|
||||
foreach (snapshot; cas.getSnapshots()) {
|
||||
writeln(snapshot);
|
||||
// if (snapshot !is null) {
|
||||
// writeln(cast(string) snapshot.data);
|
||||
// snapshot.remove();
|
||||
// }
|
||||
|
||||
import std.stdio : writeln;
|
||||
|
||||
foreach (snapshot; storage.getSnapshots()) {
|
||||
writeln(cast(string) snapshot.data);
|
||||
}
|
||||
|
||||
// writeln(cas.getVersion);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue