Skip to main content

Building a Production Stateful VIDA - Part 2

In Part 1, we established the foundational components of our production stateful VIDA: project setup, database layer with MerkleTree integration, and HTTP API for peer communication. Now we'll implement the core transaction processing logic, orchestrate the main application, and demonstrate how to run the complete system.

Continuing the Build Process​

4. Transaction Processing​

Implement the core application that subscribes to VIDA transactions and processes them while maintaining state consistency.

// handler.js
import PWRJS from "@pwrjs/core";
import fetch from 'node-fetch';
import DatabaseService from './databaseService.js';

const VIDA_ID = 73746238;
const RPC_URL = "https://pwrrpc.pwrlabs.io/";
const REQUEST_TIMEOUT = 10000;

export let peersToCheckRootHashWith = [];
let pwrjsClient = null;
let subscription = null;

// Fetches the root hash from a peer node for the specified block number
async function fetchPeerRootHash(peer, blockNumber) {
    const url = `http://${peer}/rootHash?blockNumber=${blockNumber}`;
    
    try {
        const response = await fetch(url, {
            method: 'GET',
            timeout: REQUEST_TIMEOUT,
            headers: {
                'Accept': 'text/plain'
            }
        });
        
        if (response.ok) {
            const hexString = await response.text();
            const trimmed = hexString.trim();
            
            if (!trimmed) {
                console.log(`Peer ${peer} returned empty root hash for block ${blockNumber}`);
                return { success: false, rootHash: null };
            }
            
            try {
                const rootHash = Buffer.from(trimmed, 'hex');
                console.log(`Successfully fetched root hash from peer ${peer} for block ${blockNumber}`);
                return { success: true, rootHash };
            } catch (error) {
                console.log(`Invalid hex response from peer ${peer} for block ${blockNumber}`);
                return { success: false, rootHash: null };
            }
        } else {
            console.log(`Peer ${peer} returned HTTP ${response.status} for block ${blockNumber}`);
            return { success: true, rootHash: null };
        }
    } catch (error) {
        console.log(`Failed to fetch root hash from peer ${peer} for block ${blockNumber}`);
        return { success: false, rootHash: null };
    }
}

// Validates the local Merkle root against peers and persists it if a quorum of peers agree
async function checkRootHashValidityAndSave(blockNumber) {
    const localRoot = await DatabaseService.getRootHash();
    
    if (!localRoot) {
        console.log(`No local root hash available for block ${blockNumber}`);
        return;
    }
    
    let peersCount = peersToCheckRootHashWith.length;
    let quorum = Math.floor((peersCount * 2) / 3) + 1;
    let matches = 0;
    
    for (const peer of peersToCheckRootHashWith) {
        const { success, rootHash } = await fetchPeerRootHash(peer, blockNumber);
        
        if (success && rootHash) {
            if (localRoot.equals(rootHash)) {
                matches++;
            }
        } else {
            if (peersCount > 0) {
                peersCount--;
                quorum = Math.floor((peersCount * 2) / 3) + 1;
            }
        }
        
        if (matches >= quorum) {
            await DatabaseService.setBlockRootHash(blockNumber, localRoot);
            console.log(`Root hash validated and saved for block ${blockNumber}`);
            return;
        }
    }
    
    console.log(`Root hash mismatch: only ${matches}/${peersToCheckRootHashWith.length} peers agreed`);
    await DatabaseService.revertUnsavedChanges();
    subscription.setLatestCheckedBlock(BigInt(await DatabaseService.getLastCheckedBlock()));
}

// Executes a token transfer described by the given JSON payload
async function handleTransfer(jsonData, senderHex) {
    const amount = BigInt(jsonData.amount || 0);
    const receiverHex = jsonData.receiver || "";
    
    if (amount <= 0 || !receiverHex) {
        console.log("Skipping invalid transfer:", jsonData);
        return;
    }

    const senderAddress = senderHex.startsWith("0x") ? senderHex.slice(2) : senderHex;
    const receiverAddress = receiverHex.startsWith("0x") ? receiverHex.slice(2) : receiverHex;
    
    const sender = Buffer.from(senderAddress, 'hex');
    const receiver = Buffer.from(receiverAddress, 'hex');
    
    const success = await DatabaseService.transfer(sender, receiver, amount);
    
    if (success) {
        console.log(`Transfer succeeded: ${amount} from ${senderHex} to ${receiverHex}`);
    } else {
        console.log(`Transfer failed (insufficient funds): ${amount} from ${senderHex} to ${receiverHex}`);
    }
}

// Processes a single VIDA transaction
function processTransaction(txn) {    
    try {
        const dataBytes = Buffer.from(txn.data, 'hex');
    
        const dataStr = dataBytes.toString('utf8');
        const jsonData = JSON.parse(dataStr);
        
        const action = jsonData.action || "";
        
        if (action.toLowerCase() === "transfer") {
            handleTransfer(jsonData, txn.sender);
        }
    } catch (error) {
        console.error("Error processing transaction:", txn.hash, error);
    }
}

// Callback invoked as blocks are processed
async function onChainProgress(blockNumber) {
    try {
        await DatabaseService.setLastCheckedBlock(blockNumber);
        await checkRootHashValidityAndSave(blockNumber);
        console.log(`Checkpoint updated to block ${blockNumber}`);
        await DatabaseService.flush();
    } catch (error) {
        console.error("Failed to update last checked block:", blockNumber, error);
    } finally {
        return null;
    }
}

// Subscribes to VIDA transactions starting from the given block
export async function subscribeAndSync(fromBlock) {
    console.log(`Starting VIDA transaction subscription from block ${fromBlock}`);

    // Initialize RPC client
    pwrjsClient = new PWRJS(RPC_URL);
    
    // Subscribe to VIDA transactions
    subscription = pwrjsClient.subscribeToVidaTransactions(
        VIDA_ID,
        BigInt(fromBlock),
        processTransaction,
        onChainProgress
    );
    console.log(`Successfully subscribed to VIDA ${VIDA_ID} transactions`);
}

5. Main Application Logic​

Implement the main application logic for your stateful VIDA. This includes initializing the database, setting up the API server, configuring peers, and starting the transaction synchronization:

// main.js
import express from 'express';
import { GET } from './api/get.js';
import DatabaseService from './databaseService.js';
import { subscribeAndSync, peersToCheckRootHashWith } from './handler.js';

const START_BLOCK = 1;
const PORT = 8080;

const INITIAL_BALANCES = new Map([
    [Buffer.from("c767ea1d613eefe0ce1610b18cb047881bafb829", 'hex'), 1000000000000n],
    [Buffer.from("3b4412f57828d1ceb0dbf0d460f7eb1f21fed8b4", 'hex'), 1000000000000n],
    [Buffer.from("9282d39ca205806473f4fde5bac48ca6dfb9d300", 'hex'), 1000000000000n],
    [Buffer.from("e68191b7913e72e6f1759531fbfaa089ff02308a", 'hex'), 1000000000000n],
]);

let app = null;

// Initializes peer list from arguments or defaults
function initializePeers() {
    const args = process.argv.slice(2);
    
    if (args.length > 0) {
        peersToCheckRootHashWith.length = 0;
        peersToCheckRootHashWith.push(...args);
        console.log("Using peers from args:", peersToCheckRootHashWith);
    } else {
        peersToCheckRootHashWith.length = 0;
        peersToCheckRootHashWith.push("localhost:8080");
        console.log("Using default peers:", peersToCheckRootHashWith);
    }
}

// Sets up the initial account balances when starting from a fresh database
async function initInitialBalances() {
    const lastCheckedBlock = await DatabaseService.getLastCheckedBlock();
    
    if (lastCheckedBlock === 0) {
        console.log("Setting up initial balances for fresh database");
        
        for (const [address, balance] of INITIAL_BALANCES) {
            await DatabaseService.setBalance(address, balance);
            console.log(`Set initial balance for ${address.toString('hex')}: ${balance}`);
        }
        console.log("Initial balances setup completed");
    }
}

// Start the API server in a background task
async function startApiServer() {
    app = express();
    
    GET.run(app);
    
    return new Promise((resolve, reject) => {
        const server = app.listen(PORT, '0.0.0.0', (err) => {
            if (err) {
                reject(err);
            } else {
                console.log(`Starting API server on port ${PORT}`);
                setTimeout(() => {
                    console.log(`API server started on http://0.0.0.0:${PORT}`);
                    resolve(server);
                }, 2000);
            }
        });
    });
}

// Sets up shutdown handlers for graceful shutdown
function setupShutdownHandlers() {
    const gracefulShutdown = async (signal) => {
        console.log(`Received ${signal}, shutting down gracefully...`);
        process.exit(0);
    };
    
    process.on('SIGINT', gracefulShutdown);
    process.on('SIGTERM', gracefulShutdown);
}

// Application entry point for synchronizing VIDA transactions
// with the local Merkle-backed database.
async function main() {
    console.log("Starting PWR VIDA Transaction Synchronizer...");

    initializePeers();
    await DatabaseService.initialize();
    await startApiServer();
    await initInitialBalances();

    const lastBlock = await DatabaseService.getLastCheckedBlock();
    const fromBlock = lastBlock > 0 ? lastBlock : START_BLOCK;

    console.log(`Starting synchronization from block ${fromBlock}`);

    await subscribeAndSync(fromBlock);

    // Keep the main thread alive
    console.log("Application started successfully. Press Ctrl+C to exit.");
    // Graceful shutdown
    setupShutdownHandlers();
}

main().catch(console.error);

6. Run the project​

To run the PWR Stateful VIDA project, add the following command:

node main.js

Transaction Data Structure​

Define the JSON schema for your stateful VIDA transactions:

Transfer Transaction Example:

{
    "action": "transfer",
    "receiver": "0x3b4412f57828d1ceb0dbf0d460f7eb1f21fed8b4",
    "amount": 1000000000
}

Best Practices for Stateful VIDAs​

  1. Always Validate State: Never trust local state without peer consensus
  2. Handle Consensus Failures: Implement robust error recovery mechanisms
  3. Monitor Performance: Track consensus time and database performance
  4. Secure Peer Communication: Use HTTPS in production environments
  5. Backup Strategy: Regular database backups for disaster recovery
  6. Gradual Rollouts: Test thoroughly before deploying state changes

Conclusion​

Building stateful VIDAs requires careful consideration of data consistency, consensus mechanisms, and error handling. This tutorial provides a foundation for creating robust, production-ready applications that maintain critical state on the PWR Chain.

The example token transfer system demonstrates all key concepts you'll need for more complex stateful applications, including financial systems, voting mechanisms, and other use cases where data integrity is paramount.

Remember: stateful VIDAs trade simplicity for consistency and reliability. Choose this architecture when your application requires strong guarantees about data integrity and state consistency across network participants.