401 epic improve networking layer

msb.mjs

@@ -15,8 +15,8 @@ const rpc = {
 }
 
 const options = args.includes('--rpc') ? rpc : { storeName }
-
-const msb = new MainSettlementBus(createConfig(ENV.MAINNET, options));
+const config = createConfig(ENV.MAINNET, options)
+const msb = new MainSettlementBus(config);
 
 msb.ready().then(async () => {
     if (runRpc) {
@@ -25,7 +25,7 @@ msb.ready().then(async () => {
         const port = (portIndex !== -1 && args[portIndex + 1]) ? parseInt(args[portIndex + 1], 10) : 5000;
         const hostIndex = args.indexOf('--host');
         const host = (hostIndex !== -1 && args[hostIndex + 1]) ? args[hostIndex + 1] : 'localhost';
-        startRpcServer(msb, host, port);
+        startRpcServer(msb, config , host, port);
     } else {
         console.log('RPC server will not be started.');
         msb.interactiveMode();

proto/network.proto

@@ -0,0 +1,74 @@
+syntax = "proto3";
+
+package network.v1;
+
+enum MessageType {
+    MESSAGE_TYPE_UNSPECIFIED = 0;
+    MESSAGE_TYPE_VALIDATOR_CONNECTION_REQUEST = 1;
+    MESSAGE_TYPE_VALIDATOR_CONNECTION_RESPONSE = 2;
+    MESSAGE_TYPE_LIVENESS_REQUEST = 3;
+    MESSAGE_TYPE_LIVENESS_RESPONSE = 4;
+    MESSAGE_TYPE_BROADCAST_TRANSACTION_REQUEST = 5;
+    MESSAGE_TYPE_BROADCAST_TRANSACTION_RESPONSE = 6;
+}
+
+enum ResultCode {
+    RESULT_CODE_UNSPECIFIED = 0;
+    RESULT_CODE_OK = 1;
+    RESULT_CODE_INVALID_PAYLOAD = 2;
+    RESULT_CODE_UNSUPPORTED_VERSION = 3;
+    RESULT_CODE_RATE_LIMITED = 4;
+    RESULT_CODE_TIMEOUT = 5;
+    RESULT_CODE_SIGNATURE_INVALID = 6;
+}
+
+message ValidatorConnectionRequest {
+    string issuer_address = 1;
+    bytes nonce = 2;
+    bytes signature = 3;
+}
+
+message ValidatorConnectionResponse {
+    string issuer_address = 1;
+    bytes nonce = 2;
+    bytes signature = 3;
+    ResultCode result = 4;
+}
+
+message LivenessRequest {
+    bytes nonce = 1;
+    bytes signature = 2;
+}
+
+message LivenessResponse {
+    bytes nonce = 1;
+    bytes signature = 2;
+    ResultCode result = 3;
+}
+
+message BroadcastTransactionRequest {
+    bytes data = 1; // binary encoded payload
+    bytes nonce = 2;
+    bytes signature = 3;
+}
+
+message BroadcastTransactionResponse {
+    bytes nonce = 1;
+    bytes signature = 2;
+    ResultCode result = 3;
+}
+
+message MessageHeader {
+    MessageType type = 1;
+    uint64 session_id = 2;
+    uint64 timestamp = 3;
+    oneof field {
+        ValidatorConnectionRequest            validator_connection_request  = 4;
+        ValidatorConnectionResponse           validator_connection_response  = 5;
+        LivenessRequest                       liveness_request = 6;
+        LivenessResponse                      liveness_response = 7;
+        BroadcastTransactionRequest           broadcast_transaction_request = 8;
+        BroadcastTransactionResponse          broadcast_transaction_response = 9;
+    }
+    repeated string capabilities = 10;
+}

rpc/create_server.js

@@ -3,7 +3,7 @@ import http from 'http'
 import { applyCors } from './cors.js';
 import { routes } from './routes/index.js';
 
-export const createServer = (msbInstance) => {
+export const createServer = (msbInstance, config) => {
   const server = http.createServer({}, async (req, res) => {
 
     // --- 1. Define safe 'respond' utility (Payload MUST be an object) ---
@@ -53,7 +53,7 @@ export const createServer = (msbInstance) => {
             try {
                 // This try/catch covers synchronous errors and errors from awaited promises
                 // within the route.handler function.
-                await route.handler({ req, res, respond, msbInstance });
+                await route.handler({ req, res, respond,  msbInstance, config});
             } catch (error) {
                 // Catch errors thrown directly from the handler (or its awaited parts)
                 console.error(`Error on ${route.path}:`, error);

rpc/handlers.js

@@ -52,7 +52,7 @@ export async function handleConfirmedLength({ msbInstance, respond }) {
     respond(200, { confirmed_length });
 }
 
-export async function handleBroadcastTransaction({ msbInstance, respond, req }) {
+export async function handleBroadcastTransaction({ msbInstance, config, respond, req }) {
     let body = '';
     req.on('data', chunk => {
         body += chunk.toString();
@@ -72,7 +72,7 @@ export async function handleBroadcastTransaction({ msbInstance, respond, req })
             const decodedPayload = decodeBase64Payload(payload);
             validatePayloadStructure(decodedPayload);
             const sanitizedPayload = sanitizeTransferPayload(decodedPayload);
-            const result = await broadcastTransaction(msbInstance, sanitizedPayload);
+            const result = await broadcastTransaction(msbInstance, config, sanitizedPayload);
             respond(200, { result });
         } catch (error) {
             let code = error instanceof SyntaxError ? 400 : 500;

rpc/rpc_server.js

@@ -1,8 +1,8 @@
 import { createServer } from "./create_server.js";
 
 // Called by msb.mjs file
-export function startRpcServer(msbInstance, host, port) {
-    const server = createServer(msbInstance)
+export function startRpcServer(msbInstance, config ,host, port) {
+    const server = createServer(msbInstance, config)
 
     return server.listen(port, host, () => {
         console.log(`Running RPC with http at http://${host}:${port}`);

rpc/rpc_services.js

@@ -1,6 +1,14 @@
 import { bufferToBigInt } from "../src/utils/amountSerialization.js";
-import { normalizeDecodedPayloadForJson } from "../src/utils/normalizers.js";
+import {
+    normalizeDecodedPayloadForJson,
+    normalizeTransactionOperation,
+    normalizeTransferOperation
+} from "../src/utils/normalizers.js";
 import { get_confirmed_tx_info, get_unconfirmed_tx_info } from "../src/utils/cli.js";
+import {OperationType} from "../src/utils/constants.js";
+import b4a from "b4a";
+import PartialTransaction from "../src/core/network/messaging/validators/PartialTransaction.js";
+import PartialTransfer from "../src/core/network/messaging/validators/PartialTransfer.js";
 
 export async function getBalance(msbInstance, address, confirmed) {
     const state = msbInstance.state;
@@ -36,8 +44,41 @@ export async function getUnconfirmedLength(msbInstance) {
     return msbInstance.state.getUnsignedLength();
 }
 
-export async function broadcastTransaction(msbInstance, payload) {
-    return msbInstance.broadcastTransactionCommand(payload);
+export async function broadcastTransaction(msbInstance, config, payload) {
+    if (!payload) {
+        throw new Error("Transaction payload is required for broadcasting.");
+    }
+    let normalizedPayload;
+    let isValid = false;
+    let hash;
+
+    const partialTransferValidator = new PartialTransfer(msbInstance.state, null , config);
+    const partialTransactionValidator = new PartialTransaction(msbInstance.state, null , config);
+
+    if (payload.type === OperationType.TRANSFER) {
+        normalizedPayload = normalizeTransferOperation(payload, config);
+        isValid = await partialTransferValidator.validate(normalizedPayload);
+        hash = b4a.toString(normalizedPayload.tro.tx, "hex");
+    } else if (payload.type === OperationType.TX) {
+        normalizedPayload = normalizeTransactionOperation(payload, config);
+        isValid = await partialTransactionValidator.validate(normalizedPayload);
+        hash = b4a.toString(normalizedPayload.txo.tx, "hex");
+    }
+
+    if (!isValid) {
+        throw new Error("Invalid transaction payload.");
+    }
+
+    const success = await msbInstance.broadcastPartialTransaction(payload);
+
+    if (!success) {
+        throw new Error("Failed to broadcast transaction after multiple attempts.");
+    }
+
+    const signedLength = msbInstance.state.getSignedLength();
+    const unsignedLength = msbInstance.state.getUnsignedLength();
+
+    return { message: "Transaction broadcasted successfully.", signedLength, unsignedLength, tx: hash };
 }
 
 export async function getTxHashes(msbInstance, start, end) {

rpc/utils/helpers.js

@@ -1,5 +1,5 @@
 import b4a from "b4a"
-import { operationToPayload } from "../../src/utils/operations.js"
+import { operationToPayload } from "../../src/utils/applyOperations.js"
 export function decodeBase64Payload(base64) {
 	let decodedPayloadString
 	try {

src/core/network/identity/NetworkWalletFactory.js

@@ -1,7 +1,7 @@
 import PeerWallet from 'trac-wallet';
 import b4a from 'b4a';
 
-export class NetworkWalletFactory {
+class NetworkWalletFactory {
     static provide(options = {}) {
         const {
             enableWallet,
@@ -28,7 +28,7 @@ export class NetworkWalletFactory {
 // TODO: Once Wallet class in trac-wallet exposes a constructor/factory that accepts an existing keyPair
 // (e.g. Wallet.fromKeyPair({ publicKey, secretKey }, networkPrefix)), replace EphemeralWallet
 // with a thin wrapper around that functionality instead of duplicating signing/verification logic.
-class EphemeralWallet {
+export class EphemeralWallet {
     #publicKey;
     #secretKey;
     #address;

src/core/network/messaging/handlers/RoleOperationHandler.js

@@ -1,10 +1,9 @@
 import {OperationType} from '../../../../utils/constants.js';
 import PartialRoleAccess from "../validators/PartialRoleAccess.js";
-import {addressToBuffer} from "../../../state/utils/address.js";
-import CompleteStateMessageOperations
-    from "../../../../messages/completeStateMessages/CompleteStateMessageOperations.js";
-import {normalizeHex} from "../../../../utils/helpers.js";
 import BaseOperationHandler from './base/BaseOperationHandler.js';
+import {applyStateMessageFactory} from "../../../../messages/state/applyStateMessageFactory.js";
+import {safeEncodeApplyOperation} from "../../../../utils/protobuf/operationHelpers.js";
+import {normalizeRoleAccessOperation} from "../../../../utils/normalizers.js";
 
 class RoleOperationHandler extends BaseOperationHandler {
     #partialRoleAccessValidator;
@@ -24,15 +23,15 @@ class RoleOperationHandler extends BaseOperationHandler {
         this.#wallet = wallet;
         this.#config = config;
         this.#network = network;
-        this.#partialRoleAccessValidator = new PartialRoleAccess(state, wallet ,this.#config)
+        this.#partialRoleAccessValidator = new PartialRoleAccess(state, this.#wallet.address ,this.#config)
     }
 
     get partialRoleAccessValidator() {
         return this.#partialRoleAccessValidator;
     }
 
     async handleOperation(message, connection) {
-        const normalizedPartialRoleAccessPayload = this.#normalizePartialRoleAccess(message)
+        const normalizedPartialRoleAccessPayload = normalizeRoleAccessOperation(message, this.#config)
         const isValid = await this.partialRoleAccessValidator.validate(normalizedPartialRoleAccessPayload)
         let completePayload = null
         if (!isValid) {
@@ -41,35 +40,38 @@ class RoleOperationHandler extends BaseOperationHandler {
 
         switch (normalizedPartialRoleAccessPayload.type) {
             case OperationType.ADD_WRITER:
-                completePayload = await new CompleteStateMessageOperations(this.#wallet, this.#config).assembleAddWriterMessage(
-                    normalizedPartialRoleAccessPayload.address,
-                    normalizedPartialRoleAccessPayload.rao.tx,
-                    normalizedPartialRoleAccessPayload.rao.txv,
-                    normalizedPartialRoleAccessPayload.rao.iw,
-                    normalizedPartialRoleAccessPayload.rao.in,
-                    normalizedPartialRoleAccessPayload.rao.is,
-                );
+                completePayload = await applyStateMessageFactory(this.#wallet, this.#config)
+                    .buildCompleteAddWriterMessage(
+                        normalizedPartialRoleAccessPayload.address,
+                        normalizedPartialRoleAccessPayload.rao.tx,
+                        normalizedPartialRoleAccessPayload.rao.txv,
+                        normalizedPartialRoleAccessPayload.rao.iw,
+                        normalizedPartialRoleAccessPayload.rao.in,
+                        normalizedPartialRoleAccessPayload.rao.is,
+                    )
                 break;
             case OperationType.REMOVE_WRITER:
-                completePayload = await new CompleteStateMessageOperations(this.#wallet, this.#config).assembleRemoveWriterMessage(
-                    normalizedPartialRoleAccessPayload.address,
-                    normalizedPartialRoleAccessPayload.rao.tx,
-                    normalizedPartialRoleAccessPayload.rao.txv,
-                    normalizedPartialRoleAccessPayload.rao.iw,
-                    normalizedPartialRoleAccessPayload.rao.in,
-                    normalizedPartialRoleAccessPayload.rao.is,
-                );
+                completePayload = await applyStateMessageFactory(this.#wallet, this.#config)
+                    .buildCompleteRemoveWriterMessage(
+                        normalizedPartialRoleAccessPayload.address,
+                        normalizedPartialRoleAccessPayload.rao.tx,
+                        normalizedPartialRoleAccessPayload.rao.txv,
+                        normalizedPartialRoleAccessPayload.rao.iw,
+                        normalizedPartialRoleAccessPayload.rao.in,
+                        normalizedPartialRoleAccessPayload.rao.is,
+                    )
                 break;
             case OperationType.ADMIN_RECOVERY:
-                completePayload = await new CompleteStateMessageOperations(this.#wallet, this.#config).assembleAdminRecoveryMessage(
-                    normalizedPartialRoleAccessPayload.address,
-                    normalizedPartialRoleAccessPayload.rao.tx,
-                    normalizedPartialRoleAccessPayload.rao.txv,
-                    normalizedPartialRoleAccessPayload.rao.iw,
-                    normalizedPartialRoleAccessPayload.rao.in,
-                    normalizedPartialRoleAccessPayload.rao.is,
-                );
-                console.log("Assembled complete role access operation:", completePayload);
+
+                completePayload = await applyStateMessageFactory(this.#wallet, this.#config)
+                    .buildCompleteAdminRecoveryMessage(
+                        normalizedPartialRoleAccessPayload.address,
+                        normalizedPartialRoleAccessPayload.rao.tx,
+                        normalizedPartialRoleAccessPayload.rao.txv,
+                        normalizedPartialRoleAccessPayload.rao.iw,
+                        normalizedPartialRoleAccessPayload.rao.in,
+                        normalizedPartialRoleAccessPayload.rao.is,
+                    )
                 break;
             default:
                 throw new Error("OperationHandler: Assembling complete role access operation failed due to unsupported operation type.");
@@ -79,35 +81,7 @@ class RoleOperationHandler extends BaseOperationHandler {
             throw new Error("OperationHandler: Assembling complete role access operation failed.");
         }
 
-        this.#network.transactionPoolService.addTransaction(completePayload)
-    }
-
-    #normalizePartialRoleAccess(payload) {
-        if (!payload || typeof payload !== 'object' || !payload.rao) {
-            throw new Error('Invalid payload for bootstrap deployment normalization.');
-        }
-        const {type, address, rao} = payload;
-        if (
-            !type ||
-            !address ||
-            !rao.tx || !rao.txv || !rao.iw || !rao.in || !rao.is
-        ) {
-            throw new Error('Missing required fields in bootstrap deployment payload.');
-        }
-
-        const normalizedRao = {
-            tx: normalizeHex(rao.tx),
-            txv: normalizeHex(rao.txv),
-            iw: normalizeHex(rao.iw),
-            in: normalizeHex(rao.in),
-            is: normalizeHex(rao.is)
-        };
-
-        return {
-            type,
-            address: addressToBuffer(address, this.#config.addressPrefix),
-            rao: normalizedRao
-        };
+        this.#network.transactionPoolService.addTransaction(safeEncodeApplyOperation(completePayload))
     }
 }