[indexer]: add streaming for transactions and events

[kodemartin]

Description

We would like to support streaming on APIs built on top of the Indexer database.

To this end we should

• analyze the feasibility and requirements for implementing the feature
• proceed with the implementation in iota-indexer

[sergiupopescu199]

@iotaledger/infrastructure

PostgreSQL provides a built-in message queue system through its NOTIFY and LISTEN commands. This asynchronous notification mechanism functions similarly to Tokio's mpsc channels, enabling real-time communication between database sessions.

Key Characteristics:

• Capacity: Up to 8 GB total queue size
• Message Size: Maximum 8 KB payload per notification
• Channels: Named channels allow for topic-based message routing
• Concurrency: Multiple sessions can listen to the same channel simultaneously

Warning

If no listeners are present for the channel, notifications sent via NOTIFY will be discarded immediately. Once a listener is registered with LISTEN, it will only receive notifications sent after the LISTEN command was executed - any notifications sent before the listener was registered are lost and cannot be retrieved.

Implementation Approach

This approach leverages PostgreSQL's NOTIFY/LISTEN mechanism to implement real-time event streaming for the indexer. By using database triggers and functions directly in PostgreSQL migrations, we can automatically emit notifications whenever new records are inserted into the events and transactions tables.

-- Function to notify event changes  
CREATE OR REPLACE FUNCTION notify_event_changes()
RETURNS TRIGGER AS $$
BEGIN
   
    PERFORM pg_notify('events_channel', 
        json_build_object(
            'tx_sequence_number', NEW.tx_sequence_number,
            'event_sequence_number', NEW.event_sequence_number
        )::text
    );
    
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- Function to notify transaction changes  
CREATE OR REPLACE FUNCTION notify_transaction_changes()
RETURNS TRIGGER AS $$
BEGIN
    -- Send notification with transaction sequence number
    PERFORM pg_notify('transactions_channel',
                json_build_object(
                    'tx_sequence_number', NEW.tx_sequence_number
                )::text
            );
RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- Triggers for automatic notifications
CREATE TRIGGER events_notify_trigger
    AFTER INSERT ON events
    FOR EACH ROW
    EXECUTE FUNCTION notify_event_changes();

CREATE TRIGGER transactions_notify_trigger
    AFTER INSERT ON transactions
    FOR EACH ROW
    EXECUTE FUNCTION notify_transaction_changes();

Indexer Architecture:

A dedicated Actor within the Indexer Reader will:

1. Listen: Establish persistent connections to PostgreSQL channels (transactions_channel, events_channel)
2. Process: Upon receiving notifications, execute targeted SQL queries to fetch the complete record data by leveraging the IndexerReader.
3. Stream: Forward the retrieved data to subscribers through the existing streaming infrastructure (iota_core::streamer::Streamer & iota_core::subscription_handler::SubscriptionHandler), which already contains filter capabilities for both events & transactions.

Note

Batch Processing Optimization: The Actor collects incoming notifications and processes them in batches to minimize database queries. For each batch, a single SQL query is executed to fetch all newly inserted records at once. The retrieved data is then distributed to all matching subscribers based on their individual filter criteria, ensuring efficient resource usage while maintaining real-time delivery.

Challenges

The Diesel crate does not provide NOTIFY/LISTEN implementation in its stable release, though it is available in their latest master branch. However, attempting to use the master branch introduces breaking changes that cause the existing migration code to fail due to incompatible trait bounds, it will require more investigation, It's possible that the official release could fix this, the only thing is to wait.

Manual implementation would be extremely laborious and error-prone, particularly since existing implementation on master rely heavily on unsafe code for interfacing with underlying C libraries to get notification messages. Therefore, I propose leveraging the mature, battle-tested implementation provided by the tokio-postgres crate, which offers memory-safe async NOTIFY/LISTEN functionality.

The tokio-postgres crate will be used exclusively for the notification listening component, while all SQL queries for retrieving events and transactions will continue using Diesel as the primary ORM by leveraging the IndexerReader, maintaining consistency with the existing codebase architecture. This hybrid approach provides immediate functionality while positioning us for easy migration once Diesel's stable release includes NOTIFY/LISTEN support, requiring minimal refactoring at that time.

[sergiupopescu199]

After further internal discussion, we have refined the requirements for this feature.

Although we explored other alternatives such as external message queues, we decided to go with the simplest route by leveraging the existing infrastructure. This approach minimizes the maintenance overhead and configuration complexity of external solutions, as their capabilities would be overkill for our use case.

Implementation Plan

Database Notification System

• We will use tokio-postgres until Diesel stabilizes native notification support
• The iota-indexer will provide the API to open a connection and listen to a postgres channels, and fan out new events & transactions, managing its active subscribers.
• The iota-graphql-rpc will be the service though that will use the API to open the connection and subscribe to the tokio channel that best fits the use case.

Subscribers Management

• On the client side, we will limit the number of maximum subscribers to have better control over resources to not hit OOM or unexpected behaviors.
• This can be implemented using a Semaphore to control the maximum allowed number of active subscribers.

Event Ordering Guarantees

• The most important aspect to investigate and refine is maintaining order guarantees for events and transactions
• Instead of emitting a new notification on every inserted row, we should send notifications after a complete checkpoint has been committed
• This approach will guarantee global ordering and ensure data consistency for subscribers

This design prioritizes simplicity while maintaining the robustness and ordering guarantees required for reliable event streaming.