Fix LSN encoding in responses and handling in JS (#2446)

Follow-up to #2442

Randomizing the WAL start point in tests has highlighted problems with
how we handle LSNs in Javascript.

Javascript can only do max 53-bit number representations, after which it
will start doing nonsense without telling you (gotta love it), but LSNs
span the entire 64-bit range.

In order to fix that, we encode the `lsn` and `global_last_seen_lsn` as
strings in the responses rather than integers, so that Javascript can
either compare them as strings or convert them to `BigInt`s, both would
yield valid comparisons.

JSON itself does not really care, and other languages are able to parse
large integers correctly, but for the sake of portability in the current
landscape I think it's better to keep it encoded only once and as a
string, rather than e.g. twice as `lsn` and `lsn_str` - feels like
overkill.

I've kept the `op_position` as a number, because it would be really odd
for a transaction to have `> 2^32` operations in it - maybe if the need
arises or if we want consistency between `lsn` and `op_position` I can
convert those as well.

For the `MultiShapeStream`, I decided to parse the LSNs into `BigInt`s
even though we don't really do any calculations with them and we could
just compare them as strings, but I felt like this is more explicit,
more easily understood while reading, and less error prone in the
future, and the performance cost given that this is a primarily
network-heavy utility I think is negligible, but I'm happy to change
that if anyone feels strongly about this.

Author: msfstef

.changeset/wet-melons-cry.md

@@ -0,0 +1,6 @@
+---
+"@electric-sql/experimental": patch
+"@core/sync-service": patch
+---
+
+Encode LSN as string in JSON responses for correct handling of large values (>53 bits) in Javascript.

packages/experimental/src/bigint-utils.ts

@@ -0,0 +1,11 @@
+export function bigIntMax(...args: Array<bigint | number>): bigint {
+  return BigInt(args.reduce((m, e) => (e > m ? e : m)))
+}
+
+export function bigIntMin(...args: Array<bigint | number>): bigint {
+  return BigInt(args.reduce((m, e) => (e < m ? e : m)))
+}
+
+export function bigIntCompare(a: bigint, b: bigint): 1 | -1 | 0 {
+  return a > b ? 1 : a < b ? -1 : 0
+}

packages/experimental/src/multi-shape-stream.ts

@@ -1,3 +1,4 @@
+import { bigIntCompare, bigIntMax, bigIntMin } from './bigint-utils'
 import {
   ShapeStream,
   isChangeMessage,
@@ -125,8 +126,8 @@ export class MultiShapeStream<
   // We keep track of the last lsn of data and up-to-date messages for each shape
   // so that we can skip checkForUpdates if the lsn of the up-to-date message is
   // greater than the last lsn of data.
-  #lastDataLsns: { [K in keyof TShapeRows]: number }
-  #lastUpToDateLsns: { [K in keyof TShapeRows]: number }
+  #lastDataLsns: { [K in keyof TShapeRows]: bigint }
+  #lastUpToDateLsns: { [K in keyof TShapeRows]: bigint }
 
   readonly #subscribers = new Map<
     number,
@@ -155,11 +156,11 @@ export class MultiShapeStream<
       ])
     ) as { [K in keyof TShapeRows]: ShapeStream<TShapeRows[K]> }
     this.#lastDataLsns = Object.fromEntries(
-      Object.entries(shapes).map(([key]) => [key, -Infinity])
-    ) as { [K in keyof TShapeRows]: number }
+      Object.entries(shapes).map(([key]) => [key, BigInt(-1)])
+    ) as { [K in keyof TShapeRows]: bigint }
     this.#lastUpToDateLsns = Object.fromEntries(
-      Object.entries(shapes).map(([key]) => [key, -Infinity])
-    ) as { [K in keyof TShapeRows]: number }
+      Object.entries(shapes).map(([key]) => [key, BigInt(-1)])
+    ) as { [K in keyof TShapeRows]: bigint }
     if (start) this.#start()
   }
 
@@ -176,9 +177,13 @@ export class MultiShapeStream<
           // Whats the max lsn of the up-to-date messages?
           const upToDateLsns = messages
             .filter(isControlMessage)
-            .map(({ headers }) => (headers.global_last_seen_lsn as number) ?? 0)
+            .map(({ headers }) =>
+              typeof headers.global_last_seen_lsn === `string`
+                ? BigInt(headers.global_last_seen_lsn)
+                : BigInt(0)
+            )
           if (upToDateLsns.length > 0) {
-            const maxUpToDateLsn = Math.max(...upToDateLsns)
+            const maxUpToDateLsn = bigIntMax(...upToDateLsns)
             const lastMaxUpToDateLsn = this.#lastUpToDateLsns[key]
             if (maxUpToDateLsn > lastMaxUpToDateLsn) {
               this.#lastUpToDateLsns[key] = maxUpToDateLsn
@@ -188,9 +193,11 @@ export class MultiShapeStream<
           // Whats the max lsn of the data messages?
           const dataLsns = messages
             .filter(isChangeMessage)
-            .map(({ headers }) => (headers.lsn as number) ?? 0)
+            .map(({ headers }) =>
+              typeof headers.lsn === `string` ? BigInt(headers.lsn) : BigInt(0)
+            )
           if (dataLsns.length > 0) {
-            const maxDataLsn = Math.max(...dataLsns)
+            const maxDataLsn = bigIntMax(...dataLsns)
             const lastMaxDataLsn = this.#lastDataLsns[key]
             if (maxDataLsn > lastMaxDataLsn) {
               this.#lastDataLsns[key] = maxDataLsn
@@ -224,7 +231,7 @@ export class MultiShapeStream<
   }
 
   async #checkForUpdates() {
-    const maxDataLsn = Math.max(...Object.values(this.#lastDataLsns))
+    const maxDataLsn = bigIntMax(...Object.values(this.#lastDataLsns))
     const refreshPromises = this.#shapeEntries()
       .filter(([key]) => {
         // We only need to refresh shapes that have not seen an up-to-date message
@@ -374,20 +381,20 @@ export class TransactionalMultiShapeStream<
     [K: string]: Row<unknown>
   },
 > extends MultiShapeStream<TShapeRows> {
-  #changeMessages = new Map<number, MultiShapeMessage<Row<unknown>, string>[]>()
+  #changeMessages = new Map<bigint, MultiShapeMessage<Row<unknown>, string>[]>()
   #completeLsns: {
-    [K in keyof TShapeRows]: number
+    [K in keyof TShapeRows]: bigint
   }
 
   constructor(options: MultiShapeStreamOptions<TShapeRows>) {
     super(options)
     this.#completeLsns = Object.fromEntries(
-      Object.entries(options.shapes).map(([key]) => [key, -Infinity])
-    ) as { [K in keyof TShapeRows]: number }
+      Object.entries(options.shapes).map(([key]) => [key, BigInt(-1)])
+    ) as { [K in keyof TShapeRows]: bigint }
   }
 
   #getLowestCompleteLsn() {
-    return Math.min(...Object.values(this.#completeLsns))
+    return bigIntMin(...Object.values(this.#completeLsns))
   }
 
   protected async _publish(
@@ -399,7 +406,7 @@ export class TransactionalMultiShapeStream<
       (lsn) => lsn <= lowestCompleteLsn
     )
     const messagesToPublish = lsnsToPublish
-      .sort((a, b) => a - b)
+      .sort((a, b) => bigIntCompare(a, b))
       .map((lsn) =>
         this.#changeMessages.get(lsn)?.sort((a, b) => {
           const { headers: aHeaders } = a
@@ -429,7 +436,8 @@ export class TransactionalMultiShapeStream<
       const { shape, headers } = message
       if (isChangeMessage(message)) {
         // The snapshot message does not have an lsn, so we use 0
-        const lsn = typeof headers.lsn === `number` ? headers.lsn : 0
+        const lsn =
+          typeof headers.lsn === `string` ? BigInt(headers.lsn) : BigInt(0)
         if (!this.#changeMessages.has(lsn)) {
           this.#changeMessages.set(lsn, [])
         }
@@ -439,16 +447,16 @@ export class TransactionalMultiShapeStream<
           typeof headers.last === `boolean` &&
           headers.last === true
         ) {
-          this.#completeLsns[shape] = Math.max(this.#completeLsns[shape], lsn)
+          this.#completeLsns[shape] = bigIntMax(this.#completeLsns[shape], lsn)
         }
       } else if (isControlMessage(message)) {
         if (headers.control === `up-to-date`) {
-          if (typeof headers.global_last_seen_lsn !== `number`) {
+          if (typeof headers.global_last_seen_lsn !== `string`) {
             throw new Error(`global_last_seen_lsn is not a number`)
           }
-          this.#completeLsns[shape] = Math.max(
+          this.#completeLsns[shape] = bigIntMax(
             this.#completeLsns[shape],
-            headers.global_last_seen_lsn
+            BigInt(headers.global_last_seen_lsn)
           )
         }
       }

packages/experimental/test/bigint-utils.test.ts

@@ -0,0 +1,54 @@
+import { describe, expect, it } from 'vitest'
+import { bigIntCompare, bigIntMax, bigIntMin } from '../src/bigint-utils'
+describe(`bigIntMax`, () => {
+  it(`should return the maximum of bigint and number arguments as a bigint`, () => {
+    expect(bigIntMax(BigInt(1), BigInt(2), BigInt(3))).toBe(BigInt(3))
+    expect(bigIntMax(5, 10, 2)).toBe(BigInt(10))
+    expect(bigIntMax(BigInt(1), 2, BigInt(3), 4)).toBe(BigInt(4))
+  })
+
+  it(`should return the only element as a bigint when there is one argument`, () => {
+    expect(bigIntMax(BigInt(42))).toBe(BigInt(42))
+    expect(bigIntMax(99)).toBe(BigInt(99))
+  })
+
+  it(`should handle negative numbers and bigints`, () => {
+    expect(bigIntMax(BigInt(-10), BigInt(-5), BigInt(-1))).toBe(BigInt(-1))
+    expect(bigIntMax(-100, -50, -10)).toBe(BigInt(-10))
+  })
+})
+
+describe(`bigIntMin`, () => {
+  it(`should return the minimum of bigint and number arguments as a bigint`, () => {
+    expect(bigIntMin(BigInt(1), BigInt(2), BigInt(3))).toBe(BigInt(1))
+    expect(bigIntMin(5, 10, 2)).toBe(BigInt(2))
+    expect(bigIntMin(BigInt(1), 2, BigInt(3), 4)).toBe(BigInt(1))
+  })
+
+  it(`should return the only element as a bigint when there is one argument`, () => {
+    expect(bigIntMin(BigInt(42))).toBe(BigInt(42))
+    expect(bigIntMin(99)).toBe(BigInt(99))
+  })
+
+  it(`should handle negative numbers and bigints`, () => {
+    expect(bigIntMin(BigInt(-10), BigInt(-5), BigInt(-1))).toBe(BigInt(-10))
+    expect(bigIntMin(-100, -50, -10)).toBe(BigInt(-100))
+  })
+})
+
+describe(`bigIntCompare`, () => {
+  it(`should return 1 when the first bigint is greater than the second`, () => {
+    expect(bigIntCompare(BigInt(5), BigInt(3))).toBe(1)
+    expect(bigIntCompare(BigInt(100), BigInt(99))).toBe(1)
+  })
+
+  it(`should return -1 when the first bigint is less than the second`, () => {
+    expect(bigIntCompare(BigInt(3), BigInt(5))).toBe(-1)
+    expect(bigIntCompare(BigInt(99), BigInt(100))).toBe(-1)
+  })
+
+  it(`should return 0 when both bigints are equal`, () => {
+    expect(bigIntCompare(BigInt(42), BigInt(42))).toBe(0)
+    expect(bigIntCompare(BigInt(0), BigInt(0))).toBe(0)
+  })
+})

packages/experimental/test/multi-shape-stream.test.ts

@@ -326,7 +326,6 @@ describe(`TransactionalMultiShapeStream`, () => {
       multiShapeStream.subscribe((msgs: MultiShapeMessages<ShapeConfig>[]) => {
         messageGroups.push(msgs)
         if (multiShapeStream.isUpToDate) {
-          console.log(`multiShapeStream.isUpToDate`)
           resolve()
         }
       })

packages/sync-service/lib/electric/log_items.ex

@@ -35,7 +35,7 @@ defmodule Electric.LogItems do
            operation: :insert,
            txids: List.wrap(txids),
            relation: Tuple.to_list(change.relation),
-           lsn: change.log_offset.tx_offset,
+           lsn: to_string(change.log_offset.tx_offset),
            op_position: change.log_offset.op_offset
          }
        }}
@@ -52,7 +52,7 @@ defmodule Electric.LogItems do
            operation: :delete,
            txids: List.wrap(txids),
            relation: Tuple.to_list(change.relation),
-           lsn: change.log_offset.tx_offset,
+           lsn: to_string(change.log_offset.tx_offset),
            op_position: change.log_offset.op_offset
          }
        }}
@@ -69,7 +69,7 @@ defmodule Electric.LogItems do
            operation: :update,
            txids: List.wrap(txids),
            relation: Tuple.to_list(change.relation),
-           lsn: change.log_offset.tx_offset,
+           lsn: to_string(change.log_offset.tx_offset),
            op_position: change.log_offset.op_offset
          }
        }
@@ -90,7 +90,7 @@ defmodule Electric.LogItems do
            txids: List.wrap(txids),
            relation: Tuple.to_list(change.relation),
            key_change_to: change.key,
-           lsn: change.log_offset.tx_offset,
+           lsn: to_string(change.log_offset.tx_offset),
            op_position: change.log_offset.op_offset
          }
        }},
@@ -103,7 +103,7 @@ defmodule Electric.LogItems do
            txids: List.wrap(txids),
            relation: Tuple.to_list(change.relation),
            key_change_from: change.old_key,
-           lsn: new_offset.tx_offset,
+           lsn: to_string(new_offset.tx_offset),
            op_position: new_offset.op_offset
          }
        }}

packages/sync-service/lib/electric/shapes/api.ex

@@ -588,7 +588,7 @@ defmodule Electric.Shapes.Api do
   end
 
   defp up_to_date_ctl(up_to_date_lsn) do
-    %{headers: %{control: "up-to-date", global_last_seen_lsn: up_to_date_lsn}}
+    %{headers: %{control: "up-to-date", global_last_seen_lsn: to_string(up_to_date_lsn)}}
   end
 
   defp with_span(%Request{} = request, name, attributes \\ [], fun) do

packages/sync-service/test/electric/log_item_test.exs

@@ -23,7 +23,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :insert,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -40,7 +40,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :insert,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -65,7 +65,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :delete,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -90,7 +90,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :delete,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -117,7 +117,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :update,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -145,7 +145,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :update,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -171,7 +171,7 @@ defmodule Electric.LogItemsTest do
                       relation: ["public", "test"],
                       operation: :delete,
                       txids: [1],
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }}
@@ -200,7 +200,7 @@ defmodule Electric.LogItemsTest do
                       operation: :delete,
                       txids: [1],
                       key_change_to: "new_key",
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }},
@@ -213,7 +213,7 @@ defmodule Electric.LogItemsTest do
                       operation: :insert,
                       txids: [1],
                       key_change_from: "old_key",
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 1
                     }
                   }}
@@ -242,7 +242,7 @@ defmodule Electric.LogItemsTest do
                       operation: :delete,
                       txids: [1],
                       key_change_to: "new_key",
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 0
                     }
                   }},
@@ -255,7 +255,7 @@ defmodule Electric.LogItemsTest do
                       operation: :insert,
                       txids: [1],
                       key_change_from: "old_key",
-                      lsn: 0,
+                      lsn: "0",
                       op_position: 1
                     }
                   }}

packages/sync-service/test/electric/plug/router_test.exs

@@ -801,8 +801,8 @@ defmodule Electric.Plug.RouterTest do
 
       # Verify that both ops share the same tx offset and differ in their op offset by a known
       # amount.
-      op1_log_offset = LogOffset.new(op1_lsn, op1_op_position)
-      op2_log_offset = LogOffset.new(op2_lsn, op2_op_position)
+      op1_log_offset = LogOffset.new(String.to_integer(op1_lsn), op1_op_position)
+      op2_log_offset = LogOffset.new(String.to_integer(op2_lsn), op2_op_position)
 
       assert op2_log_offset == last_log_offset
 
@@ -888,8 +888,8 @@ defmodule Electric.Plug.RouterTest do
 
       # Verify that both ops share the same tx offset and differ in their op offset by a known
       # amount.
-      op1_log_offset = LogOffset.new(op1_lsn, op1_op_position)
-      op2_log_offset = LogOffset.new(op2_lsn, op2_op_position)
+      op1_log_offset = LogOffset.new(String.to_integer(op1_lsn), op1_op_position)
+      op2_log_offset = LogOffset.new(String.to_integer(op2_lsn), op2_op_position)
 
       assert op2_log_offset == last_log_offset