chore: add better logging to client proto requests/responses

clients/typescript/src/satellite/client.ts

@@ -37,6 +37,7 @@ import {
   startReplicationErrorToSatelliteError,
   shapeRequestToSatShapeReq,
   subsErrorToSatelliteError,
+  msgToString,
 } from '../util/proto'
 import { Socket, SocketFactory } from '../sockets/index'
 import _m0 from 'protobufjs/minimal.js'
@@ -83,6 +84,7 @@ import {
   UnsubscribeResponse,
 } from './shapes/types'
 import { SubscriptionsDataCache } from './shapes/cache'
+import { setMaskBit, getMaskBit } from '../util/bitmaskHelpers'
 
 type IncomingHandler = {
   handle: (
@@ -842,7 +844,8 @@ export class SatelliteClient extends EventEmitter implements Client {
   // TODO: properly handle socket errors; update connectivity state
   private handleIncoming(data: Buffer) {
     const messageOrError = toMessage(data)
-    Log.info(`Received message ${JSON.stringify(messageOrError)}`)
+    if (Log.getLevel() <= 1 && !(messageOrError instanceof SatelliteError))
+      Log.debug(`[proto] recv: ${msgToString(messageOrError)}`)
     if (messageOrError instanceof Error) {
       this.emit('error', messageOrError)
     } else {
@@ -974,7 +977,7 @@ export class SatelliteClient extends EventEmitter implements Client {
   }
 
   private sendMessage<T extends SatPbMsg>(request: T) {
-    Log.debug(`Sending message ${JSON.stringify(request)}`)
+    if (Log.getLevel() <= 1) Log.debug(`[proto] send: ${msgToString(request)}`)
     if (!this.socket) {
       throw new SatelliteError(
         SatelliteErrorCode.UNEXPECTED_STATE,
@@ -1094,55 +1097,6 @@ export function deserializeRow(
   )
 }
 
-/**
- * Sets a bit in the mask. Modifies the mask in place.
- *
- * Mask is represented as a Uint8Array, which will be serialized element-by-element as a mask.
- * This means that `indexFromStart` enumerates all bits in the mask in the order they will be serialized:
- *
- * @example
- * setMaskBit(new Uint8Array([0b00000000, 0b00000000]), 0)
- * // => new Uint8Array([0b10000000, 0b00000000])
- *
- * @example
- * setMaskBit(new Uint8Array([0b00000000, 0b00000000]), 8)
- * // => new Uint8Array([0b00000000, 0b10000000])
- *
- * @param array Uint8Array mask
- * @param indexFromStart bit index in the mask
- */
-function setMaskBit(array: Uint8Array, indexFromStart: number): void {
-  const byteIndex = Math.floor(indexFromStart / 8)
-  const bitIndex = 7 - (indexFromStart % 8)
-
-  const mask = 0x01 << bitIndex
-  array[byteIndex] = array[byteIndex] | mask
-}
-
-/**
- * Reads a bit in the mask
- *
- * Mask is represented as a Uint8Array, which will be serialized element-by-element as a mask.
- * This means that `indexFromStart` enumerates all bits in the mask in the order they will be serialized:
- *
- * @example
- * getMaskBit(new Uint8Array([0b10000000, 0b00000000]), 0)
- * // => 1
- *
- * @example
- * getMaskBit(new Uint8Array([0b10000000, 0b00000000]), 8)
- * // => 0
- *
- * @param array Uint8Array mask
- * @param indexFromStart bit index in the mask
- */
-function getMaskBit(array: Uint8Array, indexFromStart: number): number {
-  const byteIndex = Math.floor(indexFromStart / 8)
-  const bitIndex = 7 - (indexFromStart % 8)
-
-  return (array[byteIndex] >>> bitIndex) & 0x01
-}
-
 function calculateNumBytes(column_num: number): number {
   const rem = column_num % 8
   if (rem == 0) {
@@ -1187,7 +1141,7 @@ function serializeNullData(): Uint8Array {
   return typeEncoder.text('')
 }
 
-export function toMessage(data: Uint8Array): SatPbMsg | Error {
+export function toMessage(data: Uint8Array): SatPbMsg | SatelliteError {
   const code = data[0]
   const type = getTypeFromCode(code)
   const obj = getObjFromString(type)

clients/typescript/src/util/bitmaskHelpers.ts

@@ -0,0 +1,49 @@
+/**
+ * Sets a bit in the mask. Modifies the mask in place.
+ *
+ * Mask is represented as a Uint8Array, which will be serialized element-by-element as a mask.
+ * This means that `indexFromStart` enumerates all bits in the mask in the order they will be serialized:
+ *
+ * @example
+ * setMaskBit(new Uint8Array([0b00000000, 0b00000000]), 0)
+ * // => new Uint8Array([0b10000000, 0b00000000])
+ *
+ * @example
+ * setMaskBit(new Uint8Array([0b00000000, 0b00000000]), 8)
+ * // => new Uint8Array([0b00000000, 0b10000000])
+ *
+ * @param array Uint8Array mask
+ * @param indexFromStart bit index in the mask
+ */
+
+export function setMaskBit(array: Uint8Array, indexFromStart: number): void {
+  const byteIndex = Math.floor(indexFromStart / 8)
+  const bitIndex = 7 - (indexFromStart % 8)
+
+  const mask = 1 << bitIndex
+  array[byteIndex] = array[byteIndex] | mask
+}
+/**
+ * Reads a bit in the mask
+ *
+ * Mask is represented as a Uint8Array, which will be serialized element-by-element as a mask.
+ * This means that `indexFromStart` enumerates all bits in the mask in the order they will be serialized:
+ *
+ * @example
+ * getMaskBit(new Uint8Array([0b10000000, 0b00000000]), 0)
+ * // => 1
+ *
+ * @example
+ * getMaskBit(new Uint8Array([0b10000000, 0b00000000]), 8)
+ * // => 0
+ *
+ * @param array Uint8Array mask
+ * @param indexFromStart bit index in the mask
+ */
+
+export function getMaskBit(array: Uint8Array, indexFromStart: number): 1 | 0 {
+  const byteIndex = Math.floor(indexFromStart / 8)
+  const bitIndex = 7 - (indexFromStart % 8)
+
+  return ((array[byteIndex] >>> bitIndex) & 1) as 1 | 0
+}

clients/typescript/src/util/proto.ts

@@ -2,6 +2,8 @@ import * as Pb from '../_generated/protocol/satellite'
 import * as _m0 from 'protobufjs/minimal'
 import { SatelliteError, SatelliteErrorCode } from './types'
 import { ShapeRequest } from '../satellite/shapes/types'
+import { base64, typeDecoder } from './common'
+import { getMaskBit } from './bitmaskHelpers'
 
 type GetName<T extends { $type: string }> =
   T['$type'] extends `Electric.Satellite.v1_4.${infer K}` ? K : never
@@ -269,3 +271,132 @@ export function shapeRequestToSatShapeReq(
   }
   return shapeReqs
 }
+
+export function msgToString(message: SatPbMsg): string {
+  switch (message.$type) {
+    case 'Electric.Satellite.v1_4.SatAuthReq':
+      return `#SatAuthReq{id: ${message.id}, token: ${message.token}}`
+    case 'Electric.Satellite.v1_4.SatAuthResp':
+      return `#SatAuthResp{id: ${message.id}}`
+    case 'Electric.Satellite.v1_4.SatErrorResp':
+      return `#SatErrorResp{type: ${
+        Pb.SatErrorResp_ErrorCode[message.errorType]
+      }}`
+    case 'Electric.Satellite.v1_4.SatInStartReplicationReq': {
+      const schemaVersion = message.schemaVersion
+        ? ` schema: ${message.schemaVersion},`
+        : ''
+      return `#SatInStartReplicationReq{lsn: ${base64.fromBytes(
+        message.lsn
+      )},${schemaVersion} subscriptions: [${message.subscriptionIds}]}`
+    }
+    case 'Electric.Satellite.v1_4.SatInStartReplicationResp':
+      return `#SatInStartReplicationResp{}`
+    case 'Electric.Satellite.v1_4.SatInStopReplicationReq':
+      return `#SatInStopReplicationReq{}`
+    case 'Electric.Satellite.v1_4.SatInStopReplicationResp':
+      return `#SatInStopReplicationResp{}`
+    case 'Electric.Satellite.v1_4.SatMigrationNotification':
+      return `#SatMigrationNotification{to: ${message.newSchemaVersion}, from: ${message.newSchemaVersion}}`
+    case 'Electric.Satellite.v1_4.SatPingReq':
+      return `#SatPingReq{}`
+    case 'Electric.Satellite.v1_4.SatPingResp':
+      return `#SatPingResp{lsn: ${
+        message.lsn ? base64.fromBytes(message.lsn) : 'NULL'
+      }}`
+    case 'Electric.Satellite.v1_4.SatRelation': {
+      const cols = message.columns
+        .map((x) => `${x.name}: ${x.type}${x.primaryKey ? ' PK' : ''}`)
+        .join(', ')
+      return `#SatRelation{for: ${message.schemaName}.${message.tableName}, as: ${message.relationId}, cols: [${cols}]}`
+    }
+    case 'Electric.Satellite.v1_4.SatSubsDataBegin':
+      return `#SatSubsDataBegin{id: ${
+        message.subscriptionId
+      }, lsn: ${base64.fromBytes(message.lsn)}}`
+    case 'Electric.Satellite.v1_4.SatSubsDataEnd':
+      return `#SatSubsDataEnd{}`
+    case 'Electric.Satellite.v1_4.SatShapeDataBegin':
+      return `#SatShapeDataBegin{id: ${message.requestId}}`
+    case 'Electric.Satellite.v1_4.SatShapeDataEnd':
+      return `#SatShapeDataEnd{}`
+    case 'Electric.Satellite.v1_4.SatSubsDataError': {
+      const shapeErrors = message.shapeRequestError.map(
+        (x) =>
+          `${x.requestId}: ${Pb.SatSubsDataError_ShapeReqError_Code[x.code]} (${
+            x.message
+          })`
+      )
+      const code = Pb.SatSubsDataError_Code[message.code]
+      return `#SatSubsDataError{id: ${message.subscriptionId}, code: ${code}, msg: "${message.message}", errors: [${shapeErrors}]}`
+    }
+    case 'Electric.Satellite.v1_4.SatSubsReq':
+      return `#SatSubsReq{id: ${message.subscriptionId}, shapes: ${message.shapeRequests}}`
+    case 'Electric.Satellite.v1_4.SatSubsResp': {
+      if (message.err) {
+        const shapeErrors = message.err.shapeRequestError.map(
+          (x) =>
+            `${x.requestId}: ${
+              Pb.SatSubsResp_SatSubsError_ShapeReqError_Code[x.code]
+            } (${x.message})`
+        )
+        return `#SatSubsReq{id: ${message.subscriptionId}, err: ${
+          Pb.SatSubsResp_SatSubsError_Code[message.err.code]
+        } (${message.err.message}), shapes: [${shapeErrors}]}`
+      } else {
+        return `#SatSubsReq{id: ${message.subscriptionId}}`
+      }
+    }
+    case 'Electric.Satellite.v1_4.SatUnsubsReq':
+      return `#SatUnsubsReq{ids: ${message.subscriptionIds}}`
+    case 'Electric.Satellite.v1_4.SatUnsubsResp':
+      return `#SatUnsubsResp{}`
+    case 'Electric.Satellite.v1_4.SatOpLog':
+      return `#SatOpLog{ops: [${message.ops.map(opToString).join(', ')}]}`
+  }
+}
+
+function opToString(op: Pb.SatTransOp): string {
+  if (op.begin)
+    return `#Begin{lsn: ${base64.fromBytes(
+      op.begin.lsn
+    )}, ts: ${op.begin.commitTimestamp.toString()}, isMigration: ${
+      op.begin.isMigration
+    }}`
+  if (op.commit) return `#Commit{lsn: ${base64.fromBytes(op.commit.lsn)}}`
+  if (op.insert)
+    return `#Insert{for: ${op.insert.relationId}, tags: [${
+      op.insert.tags
+    }], new: [${op.insert.rowData ? rowToString(op.insert.rowData) : ''}]}`
+  if (op.update)
+    return `#Update{for: ${op.update.relationId}, tags: [${
+      op.update.tags
+    }], new: [${
+      op.update.rowData ? rowToString(op.update.rowData) : ''
+    }], old: data: [${
+      op.update.oldRowData ? rowToString(op.update.oldRowData) : ''
+    }]}`
+  if (op.delete)
+    return `#Delete{for: ${op.delete.relationId}, tags: [${
+      op.delete.tags
+    }], old: [${
+      op.delete.oldRowData ? rowToString(op.delete.oldRowData) : ''
+    }]}`
+  if (op.migrate)
+    return `#Migrate{vsn: ${op.migrate.version}, for: ${
+      op.migrate.table?.name
+    }, stmts: [${op.migrate.stmts
+      .map((x) => x.sql.replaceAll('\n', '\\n'))
+      .join('; ')}]}`
+  return ''
+}
+
+function rowToString(row: Pb.SatOpRow): string {
+  return row.values
+    .map((x, i) =>
+      getMaskBit(row.nullsBitmask, i) == 0
+        ? JSON.stringify(typeDecoder.text(x))
+        : '∅'
+    )
+    .join(', ')
+}

e2e/tests/03.03_node_satellite_sends_and_recieves_data.lux

@@ -11,11 +11,11 @@
     [invoke migrate_items_table 20230504114018]
 
 [shell satellite_1]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     [invoke node_sync_table "items"]
 [shell satellite_2]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     [invoke node_sync_table "items"]
 

e2e/tests/03.04_node_satellite_correctly_updates_serialization_caches.lux

@@ -17,11 +17,11 @@
     !\d items
 
 [shell satellite_1]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     [invoke node_sync_table "items"]
 [shell satellite_2]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     [invoke node_sync_table "items"]
 

e2e/tests/03.05_fresh_node_satellite_receives_all_migrations_during_initial_sync.lux

@@ -38,19 +38,15 @@
 [invoke setup_client 1 "electric_1" 5133]
 
 [shell satellite_1]
-  ?Sending message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatInStartReplicationReq","lsn":\{\}
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatInStartReplicationResp"\}
+  ??[proto] send: #SatInStartReplicationReq
+  ??[proto] recv: #SatInStartReplicationResp
 
   # Verifying the initial sync. The client receives migrations corresponding to all electrified tables on the server.
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatRelation","schemaName":"public",.*"tableName":"entries"
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatOpLog","ops":\[\
-    \{"\$$type":"Electric.Satellite.v[0-9_]+.SatTransOp","begin":\{.*\}\},\
-    \{"\$$type":"Electric.Satellite.v[0-9_]+.SatTransOp","migrate":\{.*,"version":"[0-9_]+",.*,"table":\{.*,"name":"entries"
-
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatRelation","schemaName":"public",.*"tableName":"items"
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatOpLog","ops":\[\
-    \{"\$$type":"Electric.Satellite.v[0-9_]+.SatTransOp","begin":\{.*\}\},\
-    \{"\$$type":"Electric.Satellite.v[0-9_]+.SatTransOp","migrate":\{.*,"version":"[0-9_]+",.*,"table":\{.*,"name":"items"
+  ??[proto] recv: #SatRelation{for: public.entries
+  ?\[proto\] recv: #SatOpLog\{.*#Migrate\{vsn: [0-9_]+, for: entries
+
+  ??[proto] recv: #SatRelation{for: public.items
+  ?\[proto\] recv: #SatOpLog\{.*#Migrate\{vsn: [0-9_]+, for: items
 
 [cleanup]
   [invoke teardown]

e2e/tests/03.06_node_satellite_does_sync_on_subscribe.lux

@@ -10,7 +10,7 @@
     [invoke migrate_items_table 20230504114018]
 
 [shell satellite_1]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     # We shouldn't see this "hello from pg" until we actually do sync
     -(hello from pg|$fail_pattern)

e2e/tests/03.07_node_satellite_can_delete_freshly_synced_rows.lux

@@ -10,7 +10,7 @@
     [invoke migrate_items_table 20230504114018]
 
 [shell satellite_1]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     # We shouldn't see this "hello from pg" until we actually do sync
     -(hello from pg|$fail_pattern)

e2e/tests/03.08_node_satellite_can_resume_subscriptions_on_reconnect.lux

@@ -10,7 +10,7 @@
     [invoke migrate_items_table 20230504114018]
 
 [shell satellite_1]
-    ?Received message \{"\$$type":"Electric.Satellite.v\d+_\d+.SatInStartReplicationResp"\}
+    ??[proto] recv: #SatInStartReplicationResp
     [invoke node_await_table "items"]
     # We shouldn't see this "hello from pg" until we actually do sync
     -(hello from pg|$fail_pattern)

e2e/tests/03.09_fresh_node_satellite_receives_only_missing_migrations_during_initial_sync.lux

@@ -33,16 +33,14 @@
   # The client starts replication from scratch.
   ?no previous LSN, start replication from scratch
   # It passes its schema version to the server.
-  ?Sending message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatInStartReplicationReq","lsn":\{\},.*?"schemaVersion":"$migration1_vsn"
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatInStartReplicationResp"\}
+  ??[proto] send: #SatInStartReplicationReq{lsn: , schema: $migration1_vsn
+  ??[proto] recv: #SatInStartReplicationResp{}
 
   # The client receives only the second migration.
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatOpLog","ops":\[\
-    \{"\$$type":"Electric.Satellite.v[0-9_]+.SatTransOp","begin":\{.*\}\},\
-    \{"\$$type":"Electric.Satellite.v[0-9_]+.SatTransOp","migrate":\{.*,"version":"$migration2_vsn",.*,"table":\{.*,"name":"bar"
+  ?\[proto\] recv: #SatOpLog\{.*#Migrate\{vsn: $migration2_vsn, for: bar
 
   # Wait for the ping message to make sure we don't receive anything mentioning $migration1_vsn from the server.
-  ?Received message \{"\$$type":"Electric.Satellite.v[0-9_]+.SatPingReq"\}
+  ??[proto] recv: #SatPingReq{}
 
 [cleanup]
   [invoke teardown]