Merge #1666

1666: Store and load swaps from SQLite + respawn the state machine on startup r=tcharding a=tcharding

This PR is the final missing piece of the database implementation.

- Moves the metadata store to the database.
- Loads swaps on startup from the database.
- Respawns the state machines for accepted swaps.

# Highly recommend reviewing this one patch by patch!

Resolves: #1243 (the persistent storage epic - Epic!)

Co-authored-by: Tobin C. Harding <[email protected]>
Co-authored-by: Thomas Eizinger <[email protected]>

Author: 3 people

Cargo.lock

@@ -0,0 +1,25 @@
+import { createActors } from "../../../lib_sdk/create_actors";
+
+setTimeout(function() {
+    describe("cnd can be restarted", function() {
+        this.timeout(60000);
+        it("after the swap was accepted", async function() {
+            const { alice, bob } = await createActors(
+                "cnd_can_be_restarted.log"
+            );
+
+            await alice.sendRequest();
+            await bob.accept();
+
+            await alice.currentSwapIsAccepted();
+            await bob.currentSwapIsAccepted();
+
+            await alice.restart();
+            await bob.restart();
+
+            await alice.currentSwapIsAccepted();
+            await bob.currentSwapIsAccepted();
+        });
+    });
+    run();
+}, 0);

api_tests/e2e/rfc003/btc_eth/cnd_can_be_restarted.ts

@@ -1,7 +1,12 @@
+import * as fs from "fs";
+import { promisify } from "util";
 import { CndInstance } from "../lib_sdk/cnd_instance";
 import { CND_CONFIGS } from "./config";
 import { LedgerConfig } from "./ledger_runner";
 
+const unlinkAsync = promisify(fs.unlink);
+const existsAsync = promisify(fs.exists);
+
 export class CndRunner {
     private runningNodes: { [key: string]: CndInstance };
 
@@ -38,6 +43,10 @@ export class CndRunner {
                 ledgerConfig
             );
 
+            if (await existsAsync(cndconfig.dbPath)) {
+                await unlinkAsync(cndconfig.dbPath); // delete the old database for the new test
+            }
+
             await process.start();
 
             return {

api_tests/e2e/rfc003/btc_eth/does_not_persist_data.ts

@@ -15,6 +15,7 @@ export interface HttpApi {
 
 export class E2ETestActorConfig {
     public readonly seed: Uint8Array;
+    public readonly dbPath: string;
 
     constructor(
         public readonly httpApiPort: number,
@@ -25,10 +26,10 @@ export class E2ETestActorConfig {
         this.httpApiPort = httpApiPort;
         this.comitPort = comitPort;
         this.seed = new Uint8Array(Buffer.from(seed, "hex"));
+        this.dbPath = tempfile(`.${this.name}.sqlite`);
     }
 
     public generateCndConfigFile(ledgerConfig: LedgerConfig): CndConfigFile {
-        const dbPath = tempfile(`.${this.name}.sqlite`);
         return {
             http_api: {
                 socket: {
@@ -37,7 +38,7 @@ export class E2ETestActorConfig {
                 },
             },
             database: {
-                sqlite: dbPath,
+                sqlite: this.dbPath,
             },
             network: {
                 listen: [`/ip4/0.0.0.0/tcp/${this.comitPort}`],

api_tests/lib/cnd_runner.ts

@@ -10,6 +10,7 @@ import "../../lib/setup_chai";
 import { Asset, AssetKind } from "../asset";
 import { CndInstance } from "../cnd_instance";
 import { Ledger, LedgerKind } from "../ledger";
+import { sleep } from "../utils";
 import { Wallets } from "../wallets";
 import { Actors } from "./index";
 
@@ -47,7 +48,10 @@ export class Actor {
         const logger = loggerFactory(name);
         logger.level = "debug";
 
-        logger.info("Created new actor with config %s", actorConfig);
+        logger.info(
+            "Created new actor with config %s",
+            JSON.stringify(actorConfig.generateCndConfigFile(ledgerConfig))
+        );
 
         return new Actor(logger, cndInstance);
     }
@@ -183,12 +187,24 @@ export class Actor {
         await this.swap.redeem(Actor.defaultActionConfig);
     }
 
-    public async assertHasNoSwaps() {
-        this.logger.debug("Checking if we have 0 swaps");
+    public async currentSwapIsAccepted() {
+        let swapEntity;
+
+        do {
+            swapEntity = await this.swap.getEntity();
+
+            await sleep(200);
+        } while (
+            swapEntity.properties.state.communication.status !== "ACCEPTED"
+        );
+    }
+
+    public async assertHasCurrentSwap() {
+        this.logger.debug("Checking if we can fetch the current swap");
 
-        const swaps = await this.cnd.getSwaps();
+        const response = await this.cnd.fetch(this.swap.self);
 
-        expect(swaps).to.have.length(0);
+        return response;
     }
 
     public async assertSwapped() {

api_tests/lib/config.ts

@@ -7,6 +7,8 @@ description = "Reference implementation of a COMIT network daemon."
 
 [dependencies]
 anyhow = "1"
+async-std = "1"
+async-trait = "0.1"
 binary_macros = "0.6"
 bitcoin = "0.19.1"
 blockchain_contracts = "0.1"
@@ -38,6 +40,7 @@ log = { version = "0.4", features = ["serde"] }
 maplit = "1"
 mime = "0.3"
 mime_guess = "2.0"
+paste = "0.1"
 pem = "0.7"
 rand = "0.7"
 reqwest = { version = "0.9", default-features = false }
@@ -65,7 +68,6 @@ bitcoincore-rpc = "0.8.0-rc1"
 maplit = "1"
 matches = "0.1.8"
 memsocket = "0.1"
-paste = "0.1"
 pretty_env_logger = "0.3"
 quickcheck = "0.9.0"
 regex = "1.3.1"

api_tests/lib_sdk/actors/actor.ts

@@ -7,3 +7,4 @@ DROP TABLE rfc003_ethereum_bitcoin_erc20_bitcoin_request_messages;
 DROP TABLE rfc003_ethereum_bitcoin_accept_messages;
 DROP TABLE rfc003_bitcoin_ethereum_accept_messages;
 DROP TABLE rfc003_decline_messages;
+DROP TABLE rfc003_swaps;