@spec active_shapes(Electric.stack_id()) :: MapSet.t(Electric.shape_handle())

Returns the list of currently active shapes being tracked in the shape matching filters.

Adds a shape to the shape matching index in the ShapeLogCollector used for matching and sending replication stream operations.

Returns a specification to start this module under a supervisor.

See Supervisor.

Callback implementation for GenServer.handle_continue/2.

Handles a replication log events.

Should be called with operations received from the replication stream.

Uuse GenServer.call/2 here to make the event processing synchronous.

This call/3 has a timeout of :infinity because timeouts are handled at the storage layer, that is this function doesn't assume any aggregate max time for the shape consumers to actually commit the new txn to disk, instead the storage backend is responsible for determining how long a write should reasonably take and if that fails it should raise.

Handles batched shape registration updates from the RequestBatcher.

Callback implementation for GenServer.init/1.

@spec mark_as_ready(Electric.stack_id()) :: :ok

Marks the collector as ready to process operations from the replication stream.

This is typically called after the initial shape registrations have been processed.

Utility for tests, monitors the SLC process.

@spec notify_flushed(
  Electric.stack_id(),
  Electric.shape_handle(),
  Electric.Replication.LogOffset.t()
) ::
  :ok

Notifies the ShapeLogCollector that a shape's data has been flushed up to a certain offset, used to mark the overall flush progress.

Should be called by consumer processes after they flush data.

Removes a shape from the shape matching index in the ShapeLogCollector. This call succeeds before the shape is actually removed from the index.

Set process flags on the given ShapeLogCollector process.

Accepts a list of flags to set, see Process.flag/2 for valid settings.

Doesn't crash if given an invalid flag or value - instead returns the list of invalid flags.

iex> ShapeLogCollector.set_process_flags("my-stack-id", min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024)
{:ok, settings: [min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024], invalid: []}