blog: Yjs durable streams on Electric Cloud

examples/tanstack-db-expo-starter/CHANGELOG.md

@@ -1,5 +1,12 @@
 # expo-db-electric-starter
 
+## 1.0.15
+
+### Patch Changes
+
+- Updated dependencies [deb7c32]
+  - @electric-sql/client@1.5.14
+
 ## 1.0.14
 
 ### Patch Changes

examples/tanstack-db-expo-starter/package.json

@@ -1,6 +1,6 @@
 {
   "name": "expo-db-electric-starter",
-  "version": "1.0.14",
+  "version": "1.0.15",
   "main": "index.ts",
   "scripts": {
     "start": "docker compose up -d && expo start",
@@ -13,7 +13,7 @@
     "api": "tsx api/index.ts"
   },
   "dependencies": {
-    "@electric-sql/client": "1.5.13",
+    "@electric-sql/client": "1.5.14",
     "@expo/metro-runtime": "~5.0.4",
     "@tanstack/electric-db-collection": "^0.0.15",
     "@tanstack/react-db": "^0.0.27",

packages/electric-telemetry/CHANGELOG.md

@@ -1,5 +1,15 @@
 # @core/electric-telemetry
 
+## 0.1.10
+
+### Patch Changes
+
+- 0aa8c00: Extend top processes by memory metric to collect processes until the specified mem usage threshold is covered.
+
+  `ELECTRIC_TELEMETRY_TOP_PROCESS_COUNT` has been renamed to `ELECTRIC_TELEMETRY_TOP_PROCESS_LIMIT` with a new format: `count:<N>` or `mem_percent:<N>`. The old env var is still accepted as a fallback.
+
+- 0aa8c00: Group request handler processes together to see their aggregated memory usage.
+
 ## 0.1.9
 
 ### Patch Changes

packages/electric-telemetry/package.json

@@ -1,5 +1,5 @@
 {
   "name": "@core/electric-telemetry",
   "private": true,
-  "version": "0.1.9"
+  "version": "0.1.10"
 }

packages/elixir-client/CHANGELOG.md

@@ -1,5 +1,11 @@
 # @core/elixir-client
 
+## 0.9.4
+
+### Patch Changes
+
+- cb2c45e: Include required headers in client mock responses
+
 ## 0.9.3
 
 ### Patch Changes

packages/elixir-client/lib/electric/client/mock.ex

@@ -198,9 +198,11 @@ defmodule Electric.Client.Mock do
   end
 
   defp build_response(opts) do
+    opts = Keyword.put_new(opts, :next_cursor, System.unique_integer([:positive, :monotonic]))
+
     %Fetch.Response{
       status: Keyword.get(opts, :status, 200),
-      headers: headers(opts[:headers] || []),
+      headers: headers(opts),
       body: jsonify(opts[:body] || []),
       schema: Keyword.get(opts, :schema, nil),
       shape_handle: Keyword.get(opts, :shape_handle, nil),
@@ -213,12 +215,14 @@ defmodule Electric.Client.Mock do
           {:shape_handle, Client.shape_handle()}
           | {:last_offset, Client.Offset.t()}
           | {:schema, Client.schema()}
+          | {:next_cursor, Client.cursor()}
         ]) :: %{String.t() => [String.t()]}
   def headers(args) do
     %{}
     |> put_optional_header("electric-handle", args[:shape_handle])
     |> put_optional_header("electric-offset", args[:last_offset])
     |> put_optional_header("electric-schema", args[:schema], &Jason.encode!/1)
+    |> put_optional_header("electric-cursor", args[:next_cursor])
   end
 
   defp put_optional_header(headers, header, value, encoder \\ & &1)

packages/elixir-client/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@core/elixir-client",
   "private": true,
-  "version": "0.9.3",
+  "version": "0.9.4",
   "scripts": {
     "publish:hex": "../../scripts/publish_hex.sh electric_client"
   }

packages/elixir-client/test/electric/client/mock_test.exs

@@ -20,8 +20,9 @@ defmodule Electric.Client.MockTest do
       start_supervised(
         {Task,
          fn ->
-           events = Client.stream(client, ctx.shape) |> Enum.take(5)
-           send(parent, {:events, events})
+           client
+           |> Client.stream(ctx.shape)
+           |> Enum.each(fn evt -> send(parent, {:event, evt}) end)
          end}
       )
 
@@ -51,10 +52,16 @@ defmodule Electric.Client.MockTest do
         ]
       )
 
-    events =
-      receive do
-        {:events, events} -> events
-      end
+    event_stream =
+      Stream.repeatedly(fn ->
+        receive do
+          {:event, event} -> event
+        after
+          1_000 -> raise "client stream has crashed"
+        end
+      end)
+
+    events = Enum.take(event_stream, 5)
 
     assert [
              %ChangeMessage{value: %{"id" => 1111}},
@@ -63,5 +70,19 @@ defmodule Electric.Client.MockTest do
              %ChangeMessage{value: %{"id" => 4444}},
              up_to_date(1234)
            ] = events
+
+    {:ok, _request} =
+      Client.Mock.response(client,
+        status: 200,
+        schema: %{id: %{type: "int8"}},
+        last_offset: Offset.new(0, 1),
+        shape_handle: "my-shape",
+        body: [
+          Client.Mock.change(value: %{id: "5555"}),
+          Client.Mock.up_to_date(lsn: 1235)
+        ]
+      )
+
+    assert [_, up_to_date(1235)] = Enum.take(event_stream, 2)
   end
 end

packages/experimental/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @electric-sql/experimental
 
+## 6.0.14
+
+### Patch Changes
+
+- Updated dependencies [deb7c32]
+  - @electric-sql/client@1.5.14
+
 ## 6.0.13
 
 ### Patch Changes

packages/experimental/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@electric-sql/experimental",
   "description": "Experimental TypeScript features for ElectricSQL.",
-  "version": "6.0.13",
+  "version": "6.0.14",
   "author": "ElectricSQL team and contributors.",
   "bugs": {
     "url": "https://github.com/electric-sql/electric/issues"

packages/react-hooks/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @electric-sql/react
 
+## 1.0.43
+
+### Patch Changes
+
+- Updated dependencies [deb7c32]
+  - @electric-sql/client@1.5.14
+
 ## 1.0.42
 
 ### Patch Changes

packages/react-hooks/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@electric-sql/react",
   "description": "React hooks for ElectricSQL",
-  "version": "1.0.42",
+  "version": "1.0.43",
   "author": "ElectricSQL team and contributors.",
   "bugs": {
     "url": "https://github.com/electric-sql/electric/issues"

packages/sync-service/CHANGELOG.md

@@ -1,5 +1,19 @@
 # @core/sync-service
 
+## 1.4.16
+
+### Patch Changes
+
+- 6c5068a: Fix stuck flush tracker when storage flush notification arrives mid-transaction in Consumer
+- 93e5d40: Fix typo in source event name
+- 64a89a0: Fixed char(n) column values being trimmed of trailing spaces in snapshot and subset queries, causing inconsistency with values from PG replication.
+- 8919ca3: Reclassify `branch_does_not_exist` error as retryable. PlanetScale returns this
+  error transiently during cluster maintenance, and classifying it as non-retryable
+  caused sources to be permanently shut down requiring manual restart.
+- d89be52: Improve shutdown times by changing the consumer supervision strategy
+- 0af96e9: Make in-memory shape db instances isolated
+- 461576d: Add known errors for pg authorization failures
+
 ## 1.4.15
 
 ### Patch Changes

packages/sync-service/config/runtime.exs

@@ -207,6 +207,7 @@ config :electric,
   max_shapes:
     env!("ELECTRIC_MAX_SHAPES", :integer, nil) ||
       env!("ELECTRIC_EXPERIMENTAL_MAX_SHAPES", :integer, nil),
+  consumer_partitions: env!("ELECTRIC_CONSUMER_PARTITIONS", :integer, nil),
   max_concurrent_requests: max_concurrent_requests,
   # Used in telemetry
   instance_id: instance_id,

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

@@ -137,6 +137,7 @@ defmodule Electric.Application do
       max_shapes: get_env(opts, :max_shapes),
       tweaks: [
         publication_alter_debounce_ms: get_env(opts, :publication_alter_debounce_ms),
+        consumer_partitions: get_env(opts, :consumer_partitions),
         registry_partitions: get_env(opts, :process_registry_partitions),
         publication_refresh_period: get_env(opts, :publication_refresh_period),
         schema_reconciler_period: get_env(opts, :schema_reconciler_period),

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

@@ -68,6 +68,7 @@ defmodule Electric.Config do
     stack_ready_timeout: 5_000,
     send_cache_headers?: true,
     max_shapes: nil,
+    consumer_partitions: nil,
     # This value should be tuned for the hardware it's running on.
     max_concurrent_requests: %{initial: 300, existing: 10_000},
     ## Storage

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

@@ -44,8 +44,15 @@ defmodule Electric.CoreSupervisor do
     inspector = Keyword.fetch!(opts, :inspector)
     persistent_kv = Keyword.fetch!(opts, :persistent_kv)
     tweaks = Keyword.fetch!(opts, :tweaks)
+    max_shapes = Keyword.fetch!(opts, :max_shapes)
 
-    consumer_supervisor_spec = {Electric.Shapes.DynamicConsumerSupervisor, [stack_id: stack_id]}
+    consumer_supervisor_spec =
+      {Electric.Shapes.DynamicConsumerSupervisor,
+       [
+         stack_id: stack_id,
+         max_shapes: max_shapes,
+         partitions: Keyword.get(tweaks, :consumer_partitions)
+       ]}
 
     shape_cache_spec = {Electric.ShapeCache, shape_cache_opts}
 
@@ -69,8 +76,7 @@ defmodule Electric.CoreSupervisor do
        period: Keyword.get(tweaks, :schema_reconciler_period, 60_000)}
 
     expiry_manager_spec =
-      {Electric.ShapeCache.ExpiryManager,
-       max_shapes: Keyword.fetch!(opts, :max_shapes), stack_id: stack_id}
+      {Electric.ShapeCache.ExpiryManager, max_shapes: max_shapes, stack_id: stack_id}
 
     child_spec =
       Supervisor.child_spec(

packages/sync-service/lib/electric/replication/shape_log_collector.ex

@@ -465,17 +465,14 @@ defmodule Electric.Replication.ShapeLogCollector do
 
     OpenTelemetry.add_span_attributes("txn.is_dropped": true)
 
-    {:ok,
-     %{
-       state
-       | flush_tracker:
-           FlushTracker.handle_txn_fragment(
-             state.flush_tracker,
-             txn_fragment,
-             [],
-             MapSet.new()
-           )
-     }}
+    flush_tracker =
+      if txn_fragment.commit do
+        FlushTracker.handle_txn_fragment(state.flush_tracker, txn_fragment, [])
+      else
+        state.flush_tracker
+      end
+
+    {:ok, %{state | flush_tracker: flush_tracker}}
   end
 
   defp handle_txn_fragment(
@@ -578,22 +575,9 @@ defmodule Electric.Replication.ShapeLogCollector do
 
     flush_tracker =
       case event do
-        %TransactionFragment{} ->
-          shapes_with_changes =
-            for {id, frag} <- events_by_handle,
-                frag.change_count > 0,
-                not MapSet.member?(undeliverable_set, id),
-                do: id,
-                into: MapSet.new()
-
-          if event.commit, do: LsnTracker.broadcast_last_seen_lsn(state.stack_id, lsn)
-
-          FlushTracker.handle_txn_fragment(
-            flush_tracker,
-            event,
-            delivered_shapes,
-            shapes_with_changes
-          )
+        %TransactionFragment{commit: commit} when not is_nil(commit) ->
+          LsnTracker.broadcast_last_seen_lsn(state.stack_id, lsn)
+          FlushTracker.handle_txn_fragment(flush_tracker, event, delivered_shapes)
 
         _ ->
           flush_tracker