Synchronization Layer

The Synchronization Layer orchestrates the continuous delivery of database event payloads into target vector databases. It handles batch processing, operation compaction, retries, and error routing.

Module Summary

SyncManager	Coordinates database-to-vector synchronization.
DeadLetterQueue	Thread-safe Dead Letter Queue (DLQ) to hold failed events.
RetryConfig	Configuration for exponential backoff retries with jitter.

API Reference

Synchronization manager coordinating event propagation to vector store adapters.

class DeadLetterQueue(max_size=1000)

Bases: object

Thread-safe Dead Letter Queue (DLQ) to hold failed events.

Parameters:

max_size (int)

__init__(max_size=1000)

Initialises the DeadLetterQueue.

Parameters:

max_size (int) – Maximum number of events to hold in the queue. A max_size <= 0 means infinite size.

Return type:

None

put(event, exception)

Add a failed event and its causing exception to the DLQ.

If the queue is full, the event is dropped and a warning is logged.

Parameters:

• event (DebeziumEventModel) – The DebeziumEventModel that failed to sync.

• exception (Exception) – The Exception raised during synchronization.

Return type:

None

get(block=True, timeout=None)

Retrieve a failed event and exception from the DLQ.

Parameters:

• block (bool) – Whether to block until an item is available.

• timeout (float | None) – Timeout in seconds if blocking.

Returns:

A tuple of (DebeziumEventModel, Exception).

Return type:

tuple[DebeziumEventModel, Exception]

size()

Return the number of events in the DLQ.

Return type:

int

is_empty()

Check if the DLQ is empty.

Return type:

bool

class RetryConfig(max_retries=3, initial_delay=1.0, backoff_factor=2.0, jitter=True)

Bases: object

Configuration for exponential backoff retries with jitter.

Parameters:

• max_retries (int)

• initial_delay (float)

• backoff_factor (float)

• jitter (bool)

__init__(max_retries=3, initial_delay=1.0, backoff_factor=2.0, jitter=True)

Initialises RetryConfig.

Parameters:

• max_retries (int) – Maximum number of retry attempts before failure.

• initial_delay (float) – Baseline delay in seconds before the first retry.

• backoff_factor (float) – Multiplier for backoff calculation.

• jitter (bool) – If True, applies random noise to delay.

Return type:

None

class SyncManager(document_builder, vector_store_adapter, soft_delete=False, retry_config=None, dlq=None, max_workers=None)

Bases: object

Coordinates database-to-vector synchronization.

Converts DebeziumEventModels using a DocumentBuilder and propagates updates/deletes to a VectorStoreAdapter.

Parameters:

• document_builder (DocumentBuilder)

• vector_store_adapter (VectorStoreAdapter)

• soft_delete (bool)

• retry_config (RetryConfig | None)

• dlq (DeadLetterQueue | None)

• max_workers (int | None)

__init__(document_builder, vector_store_adapter, soft_delete=False, retry_config=None, dlq=None, max_workers=None)

Initialises the SyncManager.

Parameters:

• document_builder (DocumentBuilder) – Builder to map Debezium events to LangChain Documents.

• vector_store_adapter (VectorStoreAdapter) – Destination vector store backend.

• soft_delete (bool) – If True, delete operations update metadata with _is_deleted=True instead of hard-deleting the document.

• retry_config (RetryConfig | None) – Settings for transient error retries.

• dlq (DeadLetterQueue | None) – Optional DeadLetterQueue to capture failed events.

• max_workers (int | None) – Maximum number of worker threads for parallel document building.

Return type:

None

sync(event)

Synchronise a Debezium change event to the vector store.

If synchronization fails after all retry attempts, the event is redirected to the DeadLetterQueue (DLQ).

Parameters:

event (DebeziumEventModel) – The parsed, canonical Debezium event.

Return type:

None

sync_batch(events)

Synchronise a batch of Debezium change events to the vector store.

Compacts the batch by doc_id to only apply the final state of each record. If the batch synchronization fails after all retry attempts, it falls back to sequential synchronization to isolate and route failing events to the DLQ.

Parameters:

events (list[DebeziumEventModel])

Return type:

None