implement state aggregation

solarkraft/src/aggregate.ts

@@ -0,0 +1,92 @@
+/**
+ * @license
+ * [Apache-2.0](https://github.com/freespek/solarkraft/blob/main/LICENSE)
+ */
+/**
+ * A command to aggregate the full contract state from the collected transactions.
+ * This command is potentially expensive. In the long run, it makes sense to
+ * collect the states from the archival nodes. For the time being, we aggregate
+ * the states from the transactions directly, in order to evaluate the approach.
+ *
+ * We need this feature primarily for input generation. This is an experimental
+ * feature in Phase 1. We always aggregate the state, starting with the minimal
+ * available height. Obviously, there is a lot of room for improvement here.
+ *
+ * Although it is tempting to aggregate transactions directly in fetch.ts,
+ * this is the wrong approach. Horizon may give us transactions out of order,
+ * so we need to sort them by height before state aggregation.
+ *
+ * Igor Konnov, 2024
+ */
+
+import { existsSync, writeFileSync } from 'fs'
+import { join } from 'path'
+import { JSONbig } from './globals.js'
+import {
+    emptyFullState,
+    loadContractCallEntry,
+    storagePath,
+    yieldListEntriesForContract,
+} from './fetcher/storage.js'
+import { applyCallToState } from './aggregator/aggregator.js'
+
+/**
+ * Aggregate the fetched transactions to compute the full contract state.
+ * @param args the aggregator arguments
+ */
+export async function aggregate(args: any) {
+    const storageRoot = storagePath(args.home)
+    if (!existsSync(storageRoot)) {
+        console.error(`The storage is empty. Run 'solarkraft fetch'`)
+        return
+    }
+
+    const contractId = args.id
+
+    // We have to sort the entries by height. Hence, we read them first and then sort.
+    let lastEntry = undefined
+    const entries = []
+    for (const e of yieldListEntriesForContract(
+        contractId,
+        join(storageRoot, contractId)
+    )) {
+        if (e.height <= args.heightTo) {
+            entries.push(e)
+        }
+        if (lastEntry && lastEntry.height === e.height) {
+            // this should not happen on the testnet, as there is only one transaction per height
+            console.warn(
+                `Height ${e.height}: transactions ${e.txHash} and ${lastEntry.txHash} may be out of order`
+            )
+        }
+        lastEntry = e
+    }
+    // sort the entries
+    entries.sort((a, b) => a.height - b.height)
+
+    // now we can aggregate the state
+    let nentries = 0
+    let state = emptyFullState()
+    for (const entry of entries) {
+        nentries++
+        const txEntry = loadContractCallEntry(args.home, entry.txHash)
+        if (txEntry.isRight()) {
+            if (args.verbose) {
+                console.log(`Height ${entry.height}: applied ${entry.txHash}`)
+            }
+            state = applyCallToState(state, txEntry.value)
+        } else {
+            console.error(
+                `Failed to load the transaction ${entry.txHash}: ${txEntry.value}`
+            )
+            return
+        }
+    }
+
+    // save the aggregated state
+    const contents = JSONbig.stringify(state)
+    writeFileSync(args.out, contents)
+    if (args.verbose) {
+        console.log(`Aggregated ${nentries} transactions into ${args.out}`)
+    }
+}

solarkraft/src/fetcher/aggregator.ts → solarkraft/src/aggregator/aggregator.ts

@@ -15,19 +15,22 @@ import {
     FieldsMap,
     FullState,
     MultiContractStorage,
-} from './storage.js'
+} from '../fetcher/storage.js'
 
 /**
  * Apply the updates from a contract call to the state.
  * @param state the state to update
  * @param callEntry the call entry to apply
  * @returns the updated state
  */
-export function aggregate(
+export function applyCallToState(
     state: FullState,
     callEntry: ContractCallEntry
 ): FullState {
-    if (!callEntry.txSuccess) {
+    if (callEntry.txSuccess !== true) {
+        console.warn(
+            `Transaction ${callEntry.txHash} failed (${callEntry.txSuccess})`
+        )
         return state
     }
 

solarkraft/src/fetcher/storage.ts

@@ -326,6 +326,9 @@ export function loadContractFullState(
         storagePath(solarkraftHome),
         contractId
     )
+    if (!existsSync(filename)) {
+        return left(`No state found for contract ${contractId}`)
+    }
     const contents = readFileSync(filename)
     const loaded = JSONbig.parse(contents)
     return right({
@@ -361,7 +364,8 @@ export function* yieldListEntriesForContract(
         if (dirent.isDirectory() && /^[0-9]+$/.exec(dirent.name)) {
             // This directory may contain several transactions for the same height.
             const height = Number.parseInt(dirent.name)
-            for (const ledgerDirent of readdirSync(join(path, dirent.name), {
+            const dirPath = join(path, dirent.name)
+            for (const ledgerDirent of readdirSync(dirPath, {
                 withFileTypes: true,
             })) {
                 // match all storage entries, which may be reported in different cases
@@ -370,10 +374,7 @@ export function* yieldListEntriesForContract(
                 )
                 if (ledgerDirent.isFile() && matcher) {
                     const txHash = matcher[1]
-                    const filename = join(
-                        ledgerDirent.path,
-                        `entry-${txHash}.json`
-                    )
+                    const filename = join(dirPath, `entry-${txHash}.json`)
                     const contents = JSONbig.parse(
                         readFileSync(filename, 'utf-8')
                     )

solarkraft/src/main.ts

@@ -11,6 +11,7 @@ import { fetch } from './fetch.js'
 import { verify } from './verify.js'
 import { list } from './list.js'
 import { SOLARKRAFT_DEFAULT_HOME } from './globals.js'
+import { aggregate } from './aggregate.js'
 
 // The default options present in every command
 const defaultOpts = (yargs: any) =>
@@ -65,6 +66,38 @@ const fetchCmd = {
     handler: fetch,
 }
 
+// aggregate: state aggregator
+const aggregateCmd = {
+    command: ['aggregate'],
+    desc: 'aggregate the fetched Soroban transactions to compute the full contract state',
+    builder: (yargs: any) =>
+        defaultOpts(yargs)
+            .option('id', {
+                desc: 'Contract id',
+                type: 'string',
+                require: true,
+            })
+            .option('out', {
+                desc: 'The name of the file to output the state',
+                type: 'string',
+                require: false,
+                default: 'state.json',
+            })
+            .option('heightTo', {
+                desc: 'The maximum height (ledger) to aggregate up to',
+                type: 'number',
+                require: false,
+                default: Infinity,
+            })
+            .option('verbose', {
+                desc: 'Print verbose output',
+                type: 'string',
+                require: false,
+                default: false,
+            }),
+    handler: aggregate,
+}
+
 // verify: transaction verifier
 const verifyCmd = {
     command: ['verify'],
@@ -106,6 +139,7 @@ const listCmd = {
 function main() {
     return yargs(process.argv.slice(2))
         .command(fetchCmd)
+        .command(aggregateCmd)
         .command(verifyCmd)
         .command(listCmd)
         .demandCommand(1)

solarkraft/test/e2e/verify.test.ts

@@ -169,8 +169,7 @@ describe('fetches the setter contract', () => {
     }
 
     // we need this to run the loop below
-    it(`fetched ${expectedTransactions} transactions`, async function done() {
-        this.timeout(timeout)
+    it(`fetched ${expectedTransactions} transactions`, async () => {
         await waitForEntries(timeout)
         // count the entries via yieldListEntriesForContract
         let txCount = 0
@@ -188,7 +187,27 @@ describe('fetches the setter contract', () => {
             txCount === expectedTransactions,
             `expected ${expectedTransactions} transactions`
         )
-        done()
+    })
+
+    // this test is co-located with the verification test, as it also needs the fetcher storage
+    it(`aggregates ${expectedTransactions} transactions`, async () => {
+        await waitForEntries(timeout)
+        spawn(
+            'solarkraft',
+            [
+                'aggregate',
+                `--home=${solarkraftHome}`,
+                `--id=${SETTER_CONTRACT_ADDR}`,
+                `--verbose=true`,
+            ],
+            { verbose: true }
+        )
+            .wait(
+                `Aggregated ${expectedTransactions} transactions into state.json`
+            )
+            .run((err) => {
+                assert(!err, `verification error: ${err}`)
+            })
     })
 
     describe('verifies the setter contract', async () => {

solarkraft/test/unit/aggregator.test.ts

@@ -15,7 +15,7 @@ import {
     emptyMultiContractStorage,
     FullState,
 } from '../../src/fetcher/storage.js'
-import { aggregate } from '../../src/fetcher/aggregator.js'
+import { applyCallToState } from '../../src/aggregator/aggregator.js'
 
 // these are unit tests, as we are trying to minimize the number of e2e tests
 describe('State aggregation', () => {
@@ -51,7 +51,7 @@ describe('State aggregation', () => {
             ]),
         }
 
-        const nextState = aggregate(empty, callEntry)
+        const nextState = applyCallToState(empty, callEntry)
         expect(nextState.contractId).to.equal(contractId)
         expect(nextState.timestamp).to.equal(callEntry.timestamp)
         expect(nextState.height).to.equal(callEntry.height)
@@ -145,7 +145,7 @@ describe('State aggregation', () => {
             ]),
         }
 
-        const nextState = aggregate(prevState, callEntry)
+        const nextState = applyCallToState(prevState, callEntry)
         expect(nextState.contractId).to.equal(contractId)
         expect(nextState.timestamp).to.equal(callEntry.timestamp)
         expect(nextState.height).to.equal(callEntry.height)