Serialize and compress transactions

quadratic-client/src/app/quadratic-core-types/index.d.ts

@@ -30,7 +30,7 @@ export type Axis = "X" | "Y";
 export interface Instant { seconds: number, }
 export interface Duration { years: number, months: number, seconds: number, }
 export interface RunError { span: Span | null, msg: RunErrorMsg, }
-export type RunErrorMsg = { "PythonError": string } | "Spill" | { "Unimplemented": string } | "UnknownError" | { "InternalError": string } | { "Unterminated": string } | { "Expected": { expected: string, got: string | null, } } | { "Unexpected": string } | { "TooManyArguments": { func_name: string, max_arg_count: number, } } | { "MissingRequiredArgument": { func_name: string, arg_name: string, } } | "BadFunctionName" | "BadCellReference" | "BadNumber" | { "ExactArraySizeMismatch": { expected: ArraySize, got: ArraySize, } } | { "ExactArrayAxisMismatch": { axis: Axis, expected: number, got: number, } } | { "ArrayAxisMismatch": { axis: Axis, expected: number, got: number, } } | "EmptyArray" | "NonRectangularArray" | "NonLinearArray" | "ArrayTooBig" | "CircularReference" | "Overflow" | "DivideByZero" | "NegativeExponent" | "NotANumber" | "Infinity" | "IndexOutOfBounds" | "NoMatch" | "InvalidArgument";
+export type RunErrorMsg = { "PythonError": string } | "Spill" | { "Unimplemented": string } | "UnknownError" | { "InternalError": string } | { "Unterminated": string } | { "Expected": { expected: string, got: string | null, } } | { "Unexpected": string } | { "TooManyArguments": { func_name: string, max_arg_count: number, } } | { "MissingRequiredArgument": { func_name: string, arg_name: string, } } | "BadFunctionName" | "BadCellReference" | "BadNumber" | "NaN" | { "ExactArraySizeMismatch": { expected: ArraySize, got: ArraySize, } } | { "ExactArrayAxisMismatch": { axis: Axis, expected: number, got: number, } } | { "ArrayAxisMismatch": { axis: Axis, expected: number, got: number, } } | "EmptyArray" | "NonRectangularArray" | "NonLinearArray" | "ArrayTooBig" | "CircularReference" | "Overflow" | "DivideByZero" | "NegativeExponent" | "NotANumber" | "Infinity" | "IndexOutOfBounds" | "NoMatch" | "InvalidArgument";
 export interface Pos { x: bigint, y: bigint, }
 export interface Rect { min: Pos, max: Pos, }
 export interface Span { start: number, end: number, }

quadratic-client/src/app/web-workers/multiplayerWebWorker/multiplayerCoreMessages.ts

@@ -7,13 +7,13 @@ export interface MultiplayerCoreSequenceNum {
 
 export interface CoreMultiplayerTransaction {
   type: 'coreMultiplayerTransaction';
-  operations: string;
+  operations: ArrayBuffer;
   transaction_id: string;
 }
 
 export interface MultiplayerCoreReceiveTransactions {
   type: 'multiplayerCoreReceiveTransactions';
-  transactions: string;
+  transactions: ReceiveTransaction[];
 }
 
 export interface MultiplayerCoreReceiveTransaction {

quadratic-client/src/app/web-workers/multiplayerWebWorker/multiplayerTypes.ts

@@ -90,7 +90,7 @@ export interface ReceiveTransaction {
   type: 'Transaction';
   id: string;
   file_id: string;
-  operations: string;
+  operations: string | Buffer;
   sequence_num: number;
 }
 
@@ -111,7 +111,7 @@ export interface SendGetTransactions {
 
 export interface ReceiveTransactions {
   type: 'Transactions';
-  transactions: string;
+  transactions: ReceiveTransaction[];
 }
 
 export interface Heartbeat {

quadratic-client/src/app/web-workers/multiplayerWebWorker/worker/multiplayerCore.ts

@@ -4,7 +4,7 @@
 
 import { debugWebWorkersMessages } from '@/app/debugFlags';
 import { CoreMultiplayerMessage, MultiplayerCoreMessage } from '../multiplayerCoreMessages';
-import { ReceiveTransaction } from '../multiplayerTypes';
+import { ReceiveTransaction, ReceiveTransactions } from '../multiplayerTypes';
 import { multiplayerServer } from './multiplayerServer';
 
 class MultiplayerCore {
@@ -51,10 +51,10 @@ class MultiplayerCore {
     });
   }
 
-  receiveTransactions(transactions: string) {
+  receiveTransactions(receive_transactions: ReceiveTransactions) {
     this.send({
       type: 'multiplayerCoreReceiveTransactions',
-      transactions,
+      transactions: receive_transactions.transactions,
     });
   }
 }

quadratic-client/src/app/web-workers/multiplayerWebWorker/worker/multiplayerServer.ts

@@ -4,6 +4,7 @@
 
 import { User } from '@auth0/auth0-spa-js';
 import * as Sentry from '@sentry/react';
+import { Buffer } from 'buffer';
 import sharedConstants from '../../../../../../updateAlertVersion.json';
 import { debugShow, debugShowMultiplayer } from '../../../debugFlags';
 import { ClientMultiplayerInit, MultiplayerState } from '../multiplayerClientMessages';
@@ -244,7 +245,7 @@ export class MultiplayerServer {
         break;
 
       case 'Transactions':
-        multiplayerCore.receiveTransactions(data.transactions);
+        multiplayerCore.receiveTransactions(data);
         break;
 
       case 'EnterRoom':
@@ -304,7 +305,7 @@ export class MultiplayerServer {
       id: transactionMessage.transaction_id,
       session_id: this.sessionId!,
       file_id: this.fileId!,
-      operations: transactionMessage.operations,
+      operations: Buffer.from(transactionMessage.operations).toString('base64'),
     };
     this.send(message);
   }

quadratic-client/src/app/web-workers/quadraticCore/worker/core.ts

@@ -23,8 +23,12 @@ import {
   SummarizeSelectionResult,
 } from '@/app/quadratic-core-types';
 import initCore, { GridController } from '@/app/quadratic-core/quadratic_core';
-import { MultiplayerCoreReceiveTransaction } from '@/app/web-workers/multiplayerWebWorker/multiplayerCoreMessages';
+import {
+  MultiplayerCoreReceiveTransaction,
+  MultiplayerCoreReceiveTransactions,
+} from '@/app/web-workers/multiplayerWebWorker/multiplayerCoreMessages';
 import * as Sentry from '@sentry/react';
+import { Buffer } from 'buffer';
 import {
   ClientCoreFindNextColumn,
   ClientCoreFindNextRow,
@@ -286,6 +290,11 @@ class Core {
       this.clientQueue.push(async () => {
         if (!this.gridController) throw new Error('Expected gridController to be defined');
         const data = message.transaction;
+
+        if (typeof data.operations === 'string') {
+          data.operations = Buffer.from(data.operations, 'base64');
+        }
+
         this.gridController.multiplayerTransaction(data.id, data.sequence_num, data.operations);
         offline.markTransactionSent(data.id);
         if (await offline.unsentTransactionsCount()) {
@@ -298,11 +307,25 @@ class Core {
     });
   }
 
-  receiveTransactions(transactions: string) {
+  receiveTransactions(receive_transactions: MultiplayerCoreReceiveTransactions) {
     return new Promise((resolve) => {
       this.clientQueue.push(async () => {
         if (!this.gridController) throw new Error('Expected gridController to be defined');
-        this.gridController.receiveMultiplayerTransactions(transactions);
+
+        for (let data of receive_transactions.transactions) {
+          // convert the base64 encoded string of operations into buffers
+          if (typeof data.operations === 'string') {
+            data.operations = Buffer.from(data.operations, 'base64');
+          }
+
+          this.gridController.multiplayerTransaction(data.id, data.sequence_num, data.operations);
+        }
+
+        // TODO(ddimaria): re-enable 5 - 7 days after we roll out the compressed
+        // transactions PR, so that we'll know all transactions are of the same version.
+        //
+        // const transactionsBuffer = JSON.stringify(receive_transactions.transactions);
+        // this.gridController.receiveMultiplayerTransactions(transactionsBuffer);
         if (await offline.unsentTransactionsCount()) {
           coreClient.sendMultiplayerState('syncing');
         } else {

quadratic-client/src/app/web-workers/quadraticCore/worker/coreClient.ts

@@ -120,8 +120,12 @@ class CoreClient {
     if (debugWebWorkers) console.log('[coreClient] initialized.');
   }
 
-  private send(message: CoreClientMessage) {
-    self.postMessage(message);
+  private send(message: CoreClientMessage, transfer?: Transferable[]) {
+    if (transfer) {
+      self.postMessage(message, transfer);
+    } else {
+      self.postMessage(message);
+    }
   }
 
   private handleMessage = async (e: MessageEvent<ClientCoreMessage>) => {
@@ -338,7 +342,8 @@ class CoreClient {
         return;
 
       case 'clientCoreExport':
-        this.send({ type: 'coreClientExport', id: e.data.id, grid: await core.export() });
+        const exported = await core.export();
+        this.send({ type: 'coreClientExport', id: e.data.id, grid: exported }, [exported.buffer]);
         return;
 
       case 'clientCoreSearch':

quadratic-client/src/app/web-workers/quadraticCore/worker/coreMultiplayer.ts

@@ -4,7 +4,7 @@ import { core } from './core';
 
 declare var self: WorkerGlobalScope &
   typeof globalThis & {
-    sendTransaction: (transactionId: string, operations: string) => void;
+    sendTransaction: (transactionId: string, operations: Uint8Array) => void;
     requestTransactions: (sequenceNum: number) => void;
   };
 
@@ -18,9 +18,13 @@ class CoreMultiplayer {
     if (debugWebWorkers) console.log('[coreMultiplayer] initialized');
   }
 
-  private send(message: CoreMultiplayerMessage) {
+  private send(message: CoreMultiplayerMessage, transfer?: Transferable[]) {
     if (!this.coreMessagePort) throw new Error('Expected coreMessagePort to be defined in CoreMultiplayer');
-    this.coreMessagePort.postMessage(message);
+    if (transfer) {
+      this.coreMessagePort.postMessage(message, transfer);
+    } else {
+      this.coreMessagePort.postMessage(message);
+    }
   }
 
   private handleMessage = (e: MessageEvent<MultiplayerCoreMessage>) => {
@@ -36,20 +40,23 @@ class CoreMultiplayer {
         break;
 
       case 'multiplayerCoreReceiveTransactions':
-        core.receiveTransactions(e.data.transactions);
+        core.receiveTransactions(e.data);
         break;
 
       default:
         console.warn('[coreMultiplayer] Unhandled message type', e.data);
     }
   };
 
-  sendTransaction = (transactionId: string, operations: string) => {
-    this.send({
-      type: 'coreMultiplayerTransaction',
-      operations,
-      transaction_id: transactionId,
-    });
+  sendTransaction = (transactionId: string, operations: ArrayBuffer) => {
+    this.send(
+      {
+        type: 'coreMultiplayerTransaction',
+        operations,
+        transaction_id: transactionId,
+      },
+      [operations]
+    );
   };
 
   requestTransactions = (sequenceNum: number) => {

quadratic-client/src/app/web-workers/quadraticCore/worker/rustCallbacks.ts

@@ -18,7 +18,7 @@ import {
 declare var self: WorkerGlobalScope &
   typeof globalThis & {
     addUnsentTransaction: (transactionId: string, transaction: string, operations: number) => void;
-    sendTransaction: (transactionId: string, operations: string) => void;
+    sendTransaction: (transactionId: string, operations: ArrayBuffer) => void;
     sendImportProgress: (
       filename: string,
       current: number,
@@ -94,8 +94,8 @@ export const addUnsentTransaction = (transactionId: string, transactions: string
   return self.addUnsentTransaction(transactionId, transactions, operations);
 };
 
-export const jsSendTransaction = (transactionId: string, operations: string) => {
-  return self.sendTransaction(transactionId, operations);
+export const jsSendTransaction = (transactionId: string, operations: Uint8Array) => {
+  return self.sendTransaction(transactionId, operations.buffer);
 };
 
 export const jsTime = (name: string) => console.time(name);

quadratic-core/src/controller/active_transactions/pending_transaction.rs

@@ -128,7 +128,8 @@ impl PendingTransaction {
             && !self.is_server()
         {
             let transaction_id = self.id.to_string();
-            match serde_json::to_string(&self.forward_operations) {
+
+            match Transaction::serialize_and_compress(&self.forward_operations) {
                 Ok(ops) => {
                     crate::wasm_bindings::js::jsSendTransaction(transaction_id, ops);
                 }