Client for sending interactions to the sequencer

package.json

@@ -44,6 +44,7 @@
     "test:integration:basic": "jest ./src/__tests__/integration/basic",
     "test:integration:basic:load": "jest --silent=false --detectOpenHandles ./src/__tests__/integration/basic/contract-loading.test.ts ",
     "test:integration:basic:arweave": "jest ./src/__tests__/integration/basic/arweave-transactions-loading",
+    "test:integration:decentralized-sequencer": "jest ./src/__tests__/integration/decentralized-sequencer --detectOpenHandles",
     "test:integration:internal-writes": "jest ./src/__tests__/integration/internal-writes",
     "test:integration:wasm": "jest ./src/__tests__/integration/wasm",
     "test:regression": "node ./node_modules/.bin/jest ./src/__tests__/regression",

src/__tests__/integration/decentralized-sequencer/interactions.ts

@@ -0,0 +1,150 @@
+import ArLocal from 'arlocal';
+import Arweave from 'arweave';
+import { JWKInterface } from 'arweave/node/lib/wallet';
+import fs from 'fs';
+import path from 'path';
+import { createServer, Server } from 'http';
+import { DeployPlugin, ArweaveSigner } from 'warp-contracts-plugin-deploy';
+import { Contract, WriteInteractionResponse } from '../../../contract/Contract';
+import { Warp } from '../../../core/Warp';
+import { WarpFactory, defaultCacheOptions, defaultWarpGwOptions } from '../../../core/WarpFactory';
+import { SourceType } from '../../../core/modules/impl/WarpGatewayInteractionsLoader';
+import { AddressInfo } from 'net';
+import { WARP_TAGS } from '../../../core/KnownTags';
+
+interface ExampleContractState {
+    counter: number;
+}
+
+// FIXME: change to the address of the sequencer on dev
+const DECENTRALIZED_SEQUENCER_URL = 'http://sequencer-0.warp.cc:1317';
+
+describe('Testing sending of interactions to a decentralized sequencer', () => {
+    let contractSrc: string;
+    let initialState: string;
+    let wallet: JWKInterface;
+    let arlocal: ArLocal;
+    let warp: Warp;
+    let contract: Contract<ExampleContractState>;
+    let sequencerServer: Server;
+    let centralizedSeqeuencerUrl: string;
+    let centralizedSequencerType: boolean;
+
+    beforeAll(async () => {
+        const port = 1813;
+        arlocal = new ArLocal(port, false);
+        await arlocal.start();
+
+        const arweave = Arweave.init({
+            host: 'localhost',
+            port: port,
+            protocol: 'http'
+        });
+
+        // a mock server simulating a centralized sequencer
+        centralizedSequencerType = false;
+        sequencerServer = createServer((req, res) => {
+            if (req.url === '/gateway/sequencer/address') {
+                res.writeHead(200, { 'Content-Type': 'application/json' });
+                res.end(JSON.stringify({
+                    url: centralizedSequencerType ? centralizedSeqeuencerUrl : DECENTRALIZED_SEQUENCER_URL,
+                    type: centralizedSequencerType ? 'centralized' : 'decentralized'
+                }));
+                return;
+            } else if (req.url === '/gateway/v2/sequencer/register') {
+                centralizedSequencerType = false;
+                res.writeHead(301, { Location: DECENTRALIZED_SEQUENCER_URL });
+                res.end();
+                return;
+            }
+            throw new Error("Unexpected sequencer path: " + req.url);
+        })
+        await new Promise<void>(resolve => {
+            sequencerServer.listen(() => {
+                const address = sequencerServer.address() as AddressInfo
+                centralizedSeqeuencerUrl = `http://localhost:${address.port}`
+                resolve()
+            })
+        })
+
+        const cacheOptions = {
+            ...defaultCacheOptions,
+            inMemory: true
+        }
+        const gatewayOptions = { ...defaultWarpGwOptions, source: SourceType.WARP_SEQUENCER, confirmationStatus: { notCorrupted: true } }
+
+        warp = WarpFactory
+            .custom(arweave, cacheOptions, 'custom')
+            .useWarpGateway(gatewayOptions, cacheOptions)
+            .build()
+            .use(new DeployPlugin());
+
+        ({ jwk: wallet } = await warp.generateWallet());
+
+        contractSrc = fs.readFileSync(path.join(__dirname, '../data/example-contract.js'), 'utf8');
+        initialState = fs.readFileSync(path.join(__dirname, '../data/example-contract-state.json'), 'utf8');
+
+        const { contractTxId } = await warp.deploy({
+            wallet: new ArweaveSigner(wallet),
+            initState: initialState,
+            src: contractSrc
+        });
+
+        contract = warp.contract<ExampleContractState>(contractTxId).setEvaluationOptions({
+            sequencerUrl: centralizedSeqeuencerUrl
+        });
+        contract.connect(wallet);
+
+    });
+
+    afterAll(async () => {
+        await arlocal.stop();
+        await new Promise(resolve => {
+            sequencerServer.close(resolve)
+        })
+    });
+
+    const getNonceFromResult = (result: WriteInteractionResponse | null): number => {
+        if (result) {
+            for (let tag of result.interactionTx.tags) {
+                if (tag.name === WARP_TAGS.SEQUENCER_NONCE) {
+                    return Number(tag.value)
+                }
+            }    
+        }
+        return -1
+    }
+
+    it('should add new interactions waiting for confirmation from the sequencer', async () => {
+        contract.setEvaluationOptions({ waitForConfirmation: true })
+
+        await contract.writeInteraction({ function: 'add' });
+        const result = await contract.writeInteraction({ function: 'add' });
+        expect(getNonceFromResult(result)).toEqual(1)
+        expect(result?.bundlrResponse).toBeUndefined();
+        expect(result?.sequencerTxHash).toBeDefined();
+    });
+
+    it('should add new interactions without waiting for confirmation from the sequencer', async () => {
+        contract.setEvaluationOptions({ waitForConfirmation: false })
+
+        await contract.writeInteraction({ function: 'add' });
+        const result = await contract.writeInteraction({ function: 'add' });
+        expect(getNonceFromResult(result)).toEqual(3)
+        expect(result?.bundlrResponse).toBeUndefined();
+        expect(result?.sequencerTxHash).toBeUndefined();
+    });
+
+    it('should follow the redirection returned by the centralized sequencer.', async () => {
+        centralizedSequencerType = true;
+        contract.setEvaluationOptions({
+            sequencerUrl: centralizedSeqeuencerUrl,
+            waitForConfirmation: true
+        });
+
+        const result = await contract.writeInteraction({ function: 'add' });
+        expect(getNonceFromResult(result)).toEqual(4)
+        expect(result?.bundlrResponse).toBeUndefined();
+        expect(result?.sequencerTxHash).toBeDefined();
+    });
+});

src/__tests__/integration/decentralized-sequencer/send-data-item.ts

@@ -0,0 +1,110 @@
+import Arweave from 'arweave';
+import { createData, DataItem, Signer } from 'warp-arbundles';
+import { ArweaveSigner } from 'warp-contracts-plugin-deploy';
+import { DecentralizedSequencerClient } from '../../../contract/sequencer/DecentralizedSequencerClient';
+import { SMART_WEAVE_TAGS, WARP_TAGS } from '../../../core/KnownTags';
+import { Tag } from '../../../utils/types/arweave-types';
+import { WarpFactory } from '../../../core/WarpFactory';
+import { WarpFetchWrapper } from '../../../core/WarpFetchWrapper';
+import { Signature } from '../../../contract/Signature';
+
+// FIXME: change to the address of the sequencer on dev
+const SEQUENCER_URL = 'http://sequencer-0.warp.cc:1317';
+
+describe('Testing a decentralized sequencer client', () => {
+  let client: DecentralizedSequencerClient;
+
+  beforeAll(async () => {
+    const warpFetchWrapper = new WarpFetchWrapper(WarpFactory.forLocal())
+    client = new DecentralizedSequencerClient(SEQUENCER_URL, warpFetchWrapper);
+  });
+
+  const createSignature = async (): Promise<Signature> => {
+    const wallet = await Arweave.crypto.generateJWK();
+    const signer = new ArweaveSigner(wallet);
+    return new Signature(WarpFactory.forLocal(), signer)
+  }
+
+  const createDataItem = async (signature: Signature, nonce: number, addNonceTag = true, addContractTag = true, signDataItem = true): Promise<DataItem> => {
+    const signer = signature.bundlerSigner;
+    const tags: Tag[] = [];
+    if (addNonceTag) {
+      tags.push(new Tag(WARP_TAGS.SEQUENCER_NONCE, String(nonce)));
+    }
+    if (addContractTag) {
+      tags.push(new Tag(SMART_WEAVE_TAGS.CONTRACT_TX_ID, "unit test contract"));
+    }
+    const dataItem = createData('some data', signer, { tags });
+    if (signDataItem) {
+      await dataItem.sign(signer);
+    }
+    return dataItem;
+  }
+
+  it('should return consecutive nonces for a given signature', async () => {
+    const signature = await createSignature()
+    let nonce = await client.getNonce(signature);
+    expect(nonce).toEqual(0);
+
+    nonce = await client.getNonce(signature);
+    expect(nonce).toEqual(1);
+  });
+
+  it('should reject a data item with an invalid nonce', async () => {
+    const signature = await createSignature()
+    const dataItem = await createDataItem(signature, 13);
+
+    expect(client.sendDataItem(dataItem, false))
+      .rejects
+      .toThrowError('account sequence mismatch, expected 0, got 13: incorrect account sequence');
+  });
+
+  it('should reject a data item without nonce', async () => {
+    const signature = await createSignature()
+    const dataItem = await createDataItem(signature, 0, false);
+
+    expect(client.sendDataItem(dataItem, true))
+      .rejects
+      .toThrowError('no sequencer nonce tag');
+  });
+
+  it('should reject a data item without contract', async () => {
+    const signature = await createSignature()
+    const dataItem = await createDataItem(signature, 0, true, false);
+
+    expect(client.sendDataItem(dataItem, true))
+      .rejects
+      .toThrowError('no contract tag');
+  });
+
+  it('should reject an unsigned data item', async () => {
+    const signature = await createSignature()
+    const dataItem = await createDataItem(signature, 0, true, true, false);
+
+    expect(client.sendDataItem(dataItem, true))
+      .rejects
+      .toThrowError('data item verification error');
+  });
+
+  it('should return a confirmed result', async () => {
+    const signature = await createSignature();
+    const nonce = await client.getNonce(signature);
+    const dataItem = await createDataItem(signature, nonce);
+    const result = await client.sendDataItem(dataItem, true);
+
+    expect(result.sequencerMoved).toEqual(false);
+    expect(result.bundlrResponse).toBeUndefined();
+    expect(result.sequencerTxHash).toBeDefined();
+  });
+
+  it('should return an unconfirmed result', async () => {
+    const signature = await createSignature();
+    const nonce = await client.getNonce(signature);
+    const dataItem = await createDataItem(signature, nonce);
+    const result = await client.sendDataItem(dataItem, false);
+
+    expect(result.sequencerMoved).toEqual(false);
+    expect(result.bundlrResponse).toBeUndefined();
+    expect(result.sequencerTxHash).toBeUndefined();
+  });
+});

src/contract/Contract.ts

@@ -12,7 +12,7 @@ import { Transaction } from '../utils/types/arweave-types';
 
 export type BenchmarkStats = { gatewayCommunication: number; stateEvaluation: number; total: number };
 
-interface BundlrResponse {
+export interface BundlrResponse {
   id: string;
   public: string;
   signature: string;
@@ -23,6 +23,7 @@ export interface WriteInteractionResponse {
   bundlrResponse?: BundlrResponse;
   originalTxId: string;
   interactionTx: Transaction | DataItem;
+  sequencerTxHash?: string;
 }
 
 export interface DREContractStatusResponse<State> {

src/contract/HandlerBasedContract.ts

@@ -20,7 +20,7 @@ import { Benchmark } from '../logging/Benchmark';
 import { LoggerFactory } from '../logging/LoggerFactory';
 import { Evolve } from '../plugins/Evolve';
 import { ArweaveWrapper } from '../utils/ArweaveWrapper';
-import { getJsonResponse, isBrowser, sleep, stripTrailingSlash } from '../utils/utils';
+import { getJsonResponse, isBrowser, sleep } from '../utils/utils';
 import {
   BenchmarkStats,
   Contract,
@@ -41,6 +41,7 @@ import { ContractInteractionState } from './states/ContractInteractionState';
 import { Crypto } from 'warp-isomorphic';
 import { VrfPluginFunctions } from '../core/WarpPlugin';
 import { createData, tagsExceedLimit, DataItem, Signer } from 'warp-arbundles';
+import { SequencerClient, createSequencerClient } from './sequencer/SequencerClient';
 
 /**
  * An implementation of {@link Contract} that is backwards compatible with current style
@@ -72,6 +73,7 @@ export class HandlerBasedContract<State> implements Contract<State> {
   private _children: HandlerBasedContract<unknown>[] = [];
   private _interactionState;
   private _dreStates = new Map<string, SortKeyCacheResult<EvalStateResult<State>>>();
+  private _sequencerClient: SequencerClient;
 
   constructor(
     private readonly _contractTxId: string,
@@ -325,7 +327,8 @@ export class HandlerBasedContract<State> implements Contract<State> {
       tags: Tags;
       strict: boolean;
       vrf: boolean;
-    }
+    },
+    sequencerRedirected = false
   ): Promise<WriteInteractionResponse | null> {
     this.logger.info('Bundle interaction input', input);
 
@@ -337,27 +340,37 @@ export class HandlerBasedContract<State> implements Contract<State> {
       options.vrf
     );
 
-    const response = this._warpFetchWrapper.fetch(
-      `${stripTrailingSlash(this._evaluationOptions.sequencerUrl)}/gateway/v2/sequencer/register`,
-      {
-        method: 'POST',
-        headers: {
-          'Content-Type': 'application/octet-stream',
-          Accept: 'application/json'
-        },
-        body: interactionDataItem.getRaw()
-      }
+    const sequencerClient = await this.getSequencerClient();
+    const sendResponse = await sequencerClient.sendDataItem(
+      interactionDataItem,
+      this._evaluationOptions.waitForConfirmation
     );
-
-    const dataItemId = await interactionDataItem.id;
+    if (sendResponse.sequencerMoved) {
+      this.logger.info(
+        `The sequencer at the given address (${this._evaluationOptions.sequencerUrl}) is redirecting to a new sequencer`
+      );
+      if (sequencerRedirected) {
+        throw new Error('Too many sequencer redirects');
+      }
+      this._sequencerClient = null;
+      return this.bundleInteraction(input, options, true);
+    }
 
     return {
-      bundlrResponse: await getJsonResponse(response),
-      originalTxId: dataItemId,
-      interactionTx: interactionDataItem
+      bundlrResponse: sendResponse.bundlrResponse,
+      originalTxId: await interactionDataItem.id,
+      interactionTx: interactionDataItem,
+      sequencerTxHash: sendResponse.sequencerTxHash
     };
   }
 
+  private async getSequencerClient(): Promise<SequencerClient> {
+    if (!this._sequencerClient) {
+      this._sequencerClient = await createSequencerClient(this._evaluationOptions.sequencerUrl, this._warpFetchWrapper);
+    }
+    return this._sequencerClient;
+  }
+
   private async createInteractionDataItem<Input>(
     input: Input,
     tags: Tags,
@@ -370,6 +383,12 @@ export class HandlerBasedContract<State> implements Contract<State> {
       await this.discoverInternalWrites<Input>(input, tags, transfer, strict, vrf);
     }
 
+    const sequencerClient = await this.getSequencerClient();
+    const nonce = await sequencerClient.getNonce(this._signature);
+    if (nonce !== undefined) {
+      tags.push(new Tag(WARP_TAGS.SEQUENCER_NONCE, String(nonce)));
+    }
+
     if (vrf) {
       tags.push(new Tag(WARP_TAGS.REQUEST_VRF, 'true'));
     }
@@ -483,6 +502,9 @@ export class HandlerBasedContract<State> implements Contract<State> {
     if (!this.isRoot()) {
       throw new Error('Evaluation options can be set only for the root contract');
     }
+    if (options.sequencerUrl) {
+      this._sequencerClient = null;
+    }
     this._evaluationOptions = {
       ...this._evaluationOptions,
       ...options