Commit 32df5ec0 authored by Éloïs's avatar Éloïs

[feat] gva prototype

parent 2f91f472
Pipeline #10027 passed with stages
in 17 minutes and 14 seconds
[alias]
bdex = "build --release --package duniter-dbex"
ca = "clippy --all"
cn = "check --manifest-path neon/native/Cargo.toml"
dex = "run --release --package duniter-dbex --"
ta = "test --all"
......
This diff is collapsed.
......@@ -33,13 +33,18 @@ members = [
"rust-bins/xtask",
"rust-libs/dubp-wot",
"rust-libs/duniter-dbs",
"rust-libs/duniter-dbs-read-ops",
"rust-libs/duniter-gva",
"rust-libs/duniter-server",
"rust-libs/tools/kv_typed",
"rust-libs/tools/kv_typed_code_gen"
]
[patch.crates-io]
dubp = { git = "https://git.duniter.org/libs/dubp-rs-libs" }
dubp-common = { git = "https://git.duniter.org/libs/dubp-rs-libs" }
#dubp = { path = "../dubp-rs-libs" }
#dubp-common = { path = "../dubp-rs-libs/common" }
#dubp-documents = { path = "../dubp-rs-libs/documents" }
#dubp-documents-parser = { path = "../dubp-rs-libs/documents-parser" }
#dubp-wallet = { path = "../dubp-rs-libs/wallet" }
#leveldb_minimal = { path = "../../../../rust/leveldb_minimal" }
......@@ -325,7 +325,10 @@ export class DuniterBlockchain {
await dal.trimSandboxes(block);
// Saves the block (DAL)
await dal.saveBlock(dbb);
await dal.saveBlock(dbb, conf);
// Send block to rust server
dal.rustServer.applyBlock(block);
// Save wot file
if (!dal.fs.isMemoryOnly()) {
......@@ -487,11 +490,16 @@ export class DuniterBlockchain {
}
static async revertBlock(
conf: ConfDTO,
number: number,
hash: string,
dal: FileDAL,
block?: DBBlock
) {
if (block && conf.gva) {
dal.rustServer.revertBlock(block.toBlockDTO());
}
const blockstamp = [number, hash].join("-");
// Revert links
......@@ -587,7 +595,7 @@ export class DuniterBlockchain {
for (const obj of block.transactions) {
obj.currency = block.currency;
let tx = TransactionDTO.fromJSONObject(obj);
await dal.saveTransaction(DBTx.fromTransactionDTO(tx));
await dal.saveTransaction(tx);
}
}
......@@ -641,7 +649,7 @@ export class DuniterBlockchain {
obj.currency = block.currency;
const tx = TransactionDTO.fromJSONObject(obj);
const txHash = tx.getHash();
await dal.removeTxByHash(txHash);
await dal.removePendingTxByHash(txHash);
}
}
......
......@@ -162,6 +162,8 @@ export const CommonConstants = {
MAXIMUM_LEN_OF_OUTPUT,
MAXIMUM_LEN_OF_UNLOCK,
PUSH_NEW_PENDING_TXS_EVERY_MS: 30_000,
POW_TURN_DURATION_PC: 100,
POW_TURN_DURATION_ARM: 500,
......
......@@ -28,8 +28,8 @@ export interface ProgramOptions {
loglevel?: string;
sqlTraces?: boolean;
memory?: boolean;
storeTxs?: boolean;
storeWw?: boolean;
gva?: boolean;
nogva?: boolean;
}
export const cliprogram: ProgramOptions = {
......
......@@ -174,6 +174,7 @@ export class BlockchainContext {
throw DataErrors[DataErrors.BLOCK_TO_REVERT_NOT_FOUND];
}
await DuniterBlockchain.revertBlock(
this.conf,
head_1.number,
head_1.hash,
this.dal,
......@@ -187,7 +188,12 @@ export class BlockchainContext {
async revertCurrentHead() {
const head_1 = await this.dal.bindexDAL.head(1);
this.logger.debug("Reverting HEAD~1... (b#%s)", head_1.number);
await DuniterBlockchain.revertBlock(head_1.number, head_1.hash, this.dal);
await DuniterBlockchain.revertBlock(
this.conf,
head_1.number,
head_1.hash,
this.dal
);
// Invalidates the head, since it has changed.
await this.refreshHead();
}
......
......@@ -43,7 +43,7 @@ import { MetaDAL } from "./sqliteDAL/MetaDAL";
import { DataErrors } from "../common-libs/errors";
import { BasicRevocableIdentity, IdentityDTO } from "../dto/IdentityDTO";
import { FileSystem } from "../system/directory";
import { Wot } from "../../../neon/lib";
import { RustDbTx, RustServer, RustServerConf, Wot } from "../../../neon/lib";
import { IIndexDAO } from "./indexDAL/abstract/IIndexDAO";
import { BIndexDAO } from "./indexDAL/abstract/BIndexDAO";
import { MIndexDAO } from "./indexDAL/abstract/MIndexDAO";
......@@ -52,7 +52,6 @@ import { CIndexDAO } from "./indexDAL/abstract/CIndexDAO";
import { IdentityForRequirements } from "../../service/BlockchainService";
import { NewLogger } from "../logger";
import { BlockchainDAO } from "./indexDAL/abstract/BlockchainDAO";
import { TxsDAO } from "./indexDAL/abstract/TxsDAO";
import { WalletDAO } from "./indexDAL/abstract/WalletDAO";
import { PeerDAO } from "./indexDAL/abstract/PeerDAO";
import { DBTx } from "../db/DBTx";
......@@ -73,7 +72,6 @@ import { LevelDBBindex } from "./indexDAL/leveldb/LevelDBBindex";
import { LevelUp } from "levelup";
import { LevelDBBlockchain } from "./indexDAL/leveldb/LevelDBBlockchain";
import { LevelDBSindex } from "./indexDAL/leveldb/LevelDBSindex";
import { SqliteTransactions } from "./indexDAL/sqlite/SqliteTransactions";
import { SqlitePeers } from "./indexDAL/sqlite/SqlitePeers";
import { LevelDBWallet } from "./indexDAL/leveldb/LevelDBWallet";
import { LevelDBCindex } from "./indexDAL/leveldb/LevelDBCindex";
......@@ -113,6 +111,9 @@ export class FileDAL implements ServerDAO {
coreFS: CFSCore;
confDAL: ConfDAO;
// Rust server
rustServer: RustServer;
// SQLite DALs
metaDAL: MetaDAL;
idtyDAL: IdentityDAL;
......@@ -121,7 +122,6 @@ export class FileDAL implements ServerDAO {
// New DAO entities
blockDAL: BlockchainDAO;
txsDAL: TxsDAO;
peerDAL: PeerDAO;
walletDAL: WalletDAO;
bindexDAL: BIndexDAO;
......@@ -164,7 +164,6 @@ export class FileDAL implements ServerDAO {
);
this.blockDAL = new LevelDBBlockchain(getLevelDB);
this.txsDAL = new SqliteTransactions(getSqliteDB);
this.peerDAL = new SqlitePeers(getSqliteDB);
this.walletDAL = new LevelDBWallet(getLevelDB);
this.bindexDAL = new LevelDBBindex(getLevelDB);
......@@ -181,7 +180,6 @@ export class FileDAL implements ServerDAO {
certDAL: this.certDAL,
msDAL: this.msDAL,
idtyDAL: this.idtyDAL,
txsDAL: this.txsDAL,
peerDAL: this.peerDAL,
confDAL: this.confDAL,
walletDAL: this.walletDAL,
......@@ -195,10 +193,15 @@ export class FileDAL implements ServerDAO {
}
async init(conf: ConfDTO) {
// Rust server
this.initRustServer(conf);
// wotb
this.wotb = this.params.wotbf();
// DALs
this.dals = [
this.blockDAL,
this.txsDAL,
this.peerDAL,
this.walletDAL,
this.bindexDAL,
......@@ -229,6 +232,21 @@ export class FileDAL implements ServerDAO {
}
}
initRustServer(conf: ConfDTO) {
let serverPubkey = conf.pair ? conf.pair.pub : null;
let rustServerConf = {
gva: conf.gva,
serverPubkey,
txsMempoolSize:
conf.txsMempoolSize || constants.SANDBOX_SIZE_TRANSACTIONS,
};
if (conf.memory) {
this.rustServer = new RustServer(rustServerConf, null);
} else {
this.rustServer = new RustServer(rustServerConf, this.rootPath);
}
}
getDBVersion() {
return this.metaDAL.getVersion();
}
......@@ -812,16 +830,30 @@ export class FileDAL implements ServerDAO {
.value();
}
getTxByHash(hash: string) {
return this.txsDAL.getTX(hash);
async getTxByHash(hash: string): Promise<DBTx | null> {
let tx = this.rustServer.getTxByHash(hash);
if (tx === null) {
return null;
} else {
let writtenBlock = tx.writtenBlock ? tx.writtenBlock : null;
let dbTx = DBTx.fromTransactionDTO(
await this.computeTxBlockstampTime(TransactionDTO.fromJSONObject(tx))
);
dbTx.block_number = writtenBlock;
return dbTx;
}
}
removeTxByHash(hash: string) {
return this.txsDAL.removeTX(hash);
removePendingTxByHash(hash: string) {
return this.rustServer.removePendingTxByHash(hash);
}
getTransactionsPending(versionMin = 0) {
return this.txsDAL.getAllPending(versionMin);
getTransactionsPending(versionMin = 0, medianTime = 0) {
return this.rustServer.getTransactionsPending(versionMin, medianTime);
}
getNewPendingTxs() {
return this.rustServer.getNewPendingTxs();
}
async getNonWritten(pubkey: string) {
......@@ -1205,12 +1237,12 @@ export class FileDAL implements ServerDAO {
}
}
async saveBlock(dbb: DBBlock) {
dbb.wrong = false;
await Promise.all([
this.saveBlockInFile(dbb),
this.saveTxsInFiles(dbb.transactions, dbb.number, dbb.medianTime),
]);
async saveBlock(block: DBBlock, conf: ConfDTO) {
block.wrong = false;
if (conf.gva) {
this.rustServer.applyBlock(block.toBlockDTO());
}
await this.saveBlockInFile(block);
}
async generateIndexes(
......@@ -1295,7 +1327,7 @@ export class FileDAL implements ServerDAO {
await this.certDAL.trimExpiredCerts(block.medianTime);
await this.msDAL.trimExpiredMemberships(block.medianTime);
await this.idtyDAL.trimExpiredIdentities(block.medianTime);
await this.txsDAL.trimExpiredNonWrittenTxs(
await this.rustServer.trimExpiredNonWrittenTxs(
block.medianTime - CommonConstants.TX_WINDOW
);
return true;
......@@ -1313,30 +1345,6 @@ export class FileDAL implements ServerDAO {
return this.writeSideFileOfBlock(block);
}
async saveTxsInFiles(
txs: TransactionDTO[],
block_number: number,
medianTime: number
) {
return Promise.all(
txs.map(async (tx) => {
const sp = tx.blockstamp.split("-");
const basedBlock = (await this.getAbsoluteBlockByNumberAndHash(
parseInt(sp[0]),
sp[1]
)) as DBBlock;
tx.blockstampTime = basedBlock.medianTime;
const txEntity = TransactionDTO.fromJSONObject(tx);
txEntity.computeAllHashes();
return this.txsDAL.addLinked(
TransactionDTO.fromJSONObject(txEntity),
block_number,
medianTime
);
})
);
}
async merkleForPeers() {
let peers = await this.listAllPeersWithStatusNewUP();
const leaves = peers.map((peer: DBPeer) => peer.hash);
......@@ -1365,8 +1373,27 @@ export class FileDAL implements ServerDAO {
return this.certDAL.saveNewCertification(cert);
}
saveTransaction(tx: DBTx) {
return this.txsDAL.addPending(tx);
saveTransaction(tx: TransactionDTO) {
return this.rustServer.addPendingTx(tx);
}
async computeTxBlockstampTime(tx: TransactionDTO): Promise<TransactionDTO> {
let blockNumber = parseInt(tx.blockstamp.split("-")[0]);
let basedBlock = await this.getBlock(blockNumber);
tx.blockstampTime = basedBlock ? basedBlock.medianTime : 0;
return tx;
}
async RustDbTxToDbTx(tx: RustDbTx): Promise<DBTx> {
let writtenBlockNumber = tx.writtenBlockNumber;
let writtenTime = tx.writtenTime;
let tx_dto = await this.computeTxBlockstampTime(
TransactionDTO.fromJSONObject(tx)
);
let db_tx = DBTx.fromTransactionDTO(tx_dto);
db_tx.block_number = writtenBlockNumber;
db_tx.time = writtenTime;
return db_tx;
}
async getTransactionsHistory(pubkey: string) {
......@@ -1374,25 +1401,36 @@ export class FileDAL implements ServerDAO {
sent: DBTx[];
received: DBTx[];
sending: DBTx[];
receiving: DBTx[];
pending: DBTx[];
} = {
sent: [],
received: [],
sending: [],
receiving: [],
pending: [],
};
const res = await Promise.all([
this.txsDAL.getLinkedWithIssuer(pubkey),
this.txsDAL.getLinkedWithRecipient(pubkey),
this.txsDAL.getPendingWithIssuer(pubkey),
this.txsDAL.getPendingWithRecipient(pubkey),
]);
history.sent = res[0] || [];
history.received = res[1] || [];
history.sending = res[2] || [];
history.pending = res[3] || [];
const res = this.rustServer.getTransactionsHistory(pubkey);
history.sent = await Promise.all(
res.sent.map(async (tx) => this.RustDbTxToDbTx(tx))
);
history.received = await Promise.all(
res.received.map(async (tx) => this.RustDbTxToDbTx(tx))
);
history.sending = await Promise.all(
res.sending.map(async (tx) => {
let tx_dto = await this.computeTxBlockstampTime(
TransactionDTO.fromJSONObject(tx)
);
return DBTx.fromTransactionDTO(tx_dto);
})
);
history.pending = await Promise.all(
res.pending.map(async (tx) => {
let tx_dto = await this.computeTxBlockstampTime(
TransactionDTO.fromJSONObject(tx)
);
return DBTx.fromTransactionDTO(tx_dto);
})
);
return history;
}
......@@ -1624,12 +1662,14 @@ export class FileDAL implements ServerDAO {
local_iindex: IindexEntry[]
): Promise<SimpleUdEntryForWallet[]> {
if (dividend) {
return this.dividendDAL.produceDividend(
let udSources = this.dividendDAL.produceDividend(
blockNumber,
dividend,
unitbase,
local_iindex
);
// TODO ESZ: call rust server: write_ud_sources(udSources: SimpleUdEntryForWallet[])
return udSources;
}
return [];
}
......
import { GenericDAO } from "./GenericDAO";
import { TransactionDTO } from "../../../dto/TransactionDTO";
import { SandBox } from "../../sqliteDAL/SandBox";
import { DBTx } from "../../../db/DBTx";
export interface TxsDAO extends GenericDAO<DBTx> {
trimExpiredNonWrittenTxs(limitTime: number): Promise<void>;
getAllPending(versionMin: number): Promise<DBTx[]>;
getTX(hash: string): Promise<DBTx>;
addLinked(
tx: TransactionDTO,
block_number: number,
time: number
): Promise<DBTx>;
addPending(dbTx: DBTx): Promise<DBTx>;
getLinkedWithIssuer(pubkey: string): Promise<DBTx[]>;
getLinkedWithRecipient(pubkey: string): Promise<DBTx[]>;
getPendingWithIssuer(pubkey: string): Promise<DBTx[]>;
getPendingWithRecipient(pubkey: string): Promise<DBTx[]>;
removeTX(hash: string): Promise<void>;
removeAll(): Promise<void>;
sandbox: SandBox<{
issuers: string[];
output_base: number;
output_amount: number;
}>;
getSandboxRoom(): Promise<number>;
setSandboxSize(size: number): void;
}
import { SQLiteDriver } from "../../drivers/SQLiteDriver";
import { MonitorExecutionTime } from "../../../debug/MonitorExecutionTime";
import { SqliteTable } from "./SqliteTable";
import { SqlNotNullableFieldDefinition } from "./SqlFieldDefinition";
import { DividendDAO, DividendEntry, UDSource } from "../abstract/DividendDAO";
import {
IindexEntry,
SimpleTxInput,
SimpleUdEntryForWallet,
SindexEntry,
} from "../../../indexer";
import { DividendDaoHandler } from "../common/DividendDaoHandler";
import { DataErrors } from "../../../common-libs/errors";
export class SqliteDividend extends SqliteTable<DividendEntry>
implements DividendDAO {
constructor(getSqliteDB: (dbName: string) => Promise<SQLiteDriver>) {
super(
"dividend",
{
pub: new SqlNotNullableFieldDefinition("VARCHAR", true, 50),
member: new SqlNotNullableFieldDefinition("BOOLEAN", true),
availables: new SqlNotNullableFieldDefinition("JSON", false),
consumed: new SqlNotNullableFieldDefinition("JSON", false),
consumedUDs: new SqlNotNullableFieldDefinition("JSON", false),
dividends: new SqlNotNullableFieldDefinition("JSON", false),
},
getSqliteDB
);
}
/**
* TECHNICAL
*/
cleanCache(): void {}
triggerInit(): void {}
/**
* INSERT
*/
@MonitorExecutionTime()
async insert(record: DividendEntry): Promise<void> {
await this.insertInTable(this.driver, record);
}
@MonitorExecutionTime()
async insertBatch(records: DividendEntry[]): Promise<void> {
if (records.length) {
return this.insertBatchInTable(this.driver, records);
}
}
private async find(sql: string, params: any[]): Promise<DividendEntry[]> {
return (await this.driver.sqlRead(sql, params)).map((r) => {
return {
pub: r.pub,
member: r.member,
availables:
r.availables == null ? null : JSON.parse(r.availables as any),
consumed: r.consumed == null ? null : JSON.parse(r.consumed as any),
consumedUDs:
r.consumedUDs == null ? null : JSON.parse(r.consumedUDs as any),
dividends: r.dividends == null ? null : JSON.parse(r.dividends as any),
};
});
}
async consume(filter: SindexEntry[]): Promise<void> {
for (const dividendToConsume of filter) {
const row = (
await this.find("SELECT * FROM dividend WHERE pub = ?", [
dividendToConsume.identifier,
])
)[0];
DividendDaoHandler.consume(row, dividendToConsume);
await this.update(
this.driver,
row,
["consumed", "consumedUDs", "availables", "dividends"],
["pub"]
);
}
}
async createMember(pub: string): Promise<void> {
const existing = (
await this.find("SELECT * FROM dividend WHERE pub = ?", [pub])
)[0];
if (!existing) {
await this.insert(DividendDaoHandler.getNewDividendEntry(pub));
} else {
await this.setMember(true, pub);
}
}
deleteMember(pub: string): Promise<void> {
return this.driver.sqlWrite("DELETE FROM dividend WHERE pub = ?", [pub]);
}
async findForDump(criterion: any): Promise<SindexEntry[]> {
return DividendDaoHandler.toDump(
await this.find("SELECT * FROM dividend", [])
);
}
findRawWithOrder(
criterion: { pub?: string },
sort: (string | (string | boolean)[])[]
): Promise<DividendEntry[]> {
let sql = `SELECT * FROM dividend ${criterion.pub ? "WHERE pub = ?" : ""}`;
if (sort.length) {
sql += ` ORDER BY ${sort
.map((s) => `${s[0]} ${s[1] ? "DESC" : "ASC"}`)
.join(", ")}`;
}
return this.find(sql, criterion.pub ? [criterion.pub] : []);
}
async findUdSourceByIdentifierPosAmountBase(
identifier: string,
pos: number,
amount: number,
base: number
): Promise<SimpleTxInput[]> {