[notify] Introduce StreamAllTransactions stream

Signed-off-by: Liran Funaro <liran.funaro@gmail.com>

Author: liran-funaro

cmd/config/app_config_test.go

@@ -71,6 +71,7 @@ func TestReadConfigSidecar(t *testing.T) {
 				MaxTimeout:         sidecar.DefaultNotificationMaxTimeout,
 				MaxActiveTxIDs:     sidecar.DefaultMaxActiveTxIDs,
 				MaxTxIDsPerRequest: sidecar.DefaultMaxTxIDsPerRequest,
+				StreamWriteTimeout: sidecar.DefaultStreamWriteTimeout,
 			},
 			LastCommittedBlockSetInterval: sidecar.DefaultLastCommittedBlockSetInterval,
 			WaitingTxsLimit:               sidecar.DefaultWaitingTxsLimit,
@@ -120,6 +121,7 @@ func TestReadConfigSidecar(t *testing.T) {
 				MaxTimeout:         10 * time.Minute,
 				MaxActiveTxIDs:     100_000,
 				MaxTxIDsPerRequest: 1000,
+				StreamWriteTimeout: 30 * time.Second,
 			},
 			LastCommittedBlockSetInterval: sidecar.DefaultLastCommittedBlockSetInterval,
 			WaitingTxsLimit:               20_000_000,

cmd/config/samples/sidecar.yaml

@@ -124,6 +124,11 @@ notification:
   max-active-tx-ids: 100_000
   # Default: 1,000
   max-tx-ids-per-request: 1000
+  # Timeout for writing notifications to the stream.
+  # If a client cannot consume notifications within this timeout, it will be disconnected
+  # to prevent blocking the notification service.
+  # Default: 30s
+  stream-write-timeout: 30s
 
 # How often the sidecar updates the coordinator with the last committed block number.
 # More frequent updates improve coordinator accuracy for dependency resolution but

cmd/config/viper.go

@@ -45,6 +45,7 @@ func NewViperWithSidecarDefaults() *viper.Viper {
 	v.SetDefault("notification.max-timeout", sidecar.DefaultNotificationMaxTimeout)
 	v.SetDefault("notification.max-active-tx-ids", sidecar.DefaultMaxActiveTxIDs)
 	v.SetDefault("notification.max-tx-ids-per-request", sidecar.DefaultMaxTxIDsPerRequest)
+	v.SetDefault("notification.stream-write-timeout", sidecar.DefaultStreamWriteTimeout)
 	v.SetDefault("last-committed-block-set-interval", sidecar.DefaultLastCommittedBlockSetInterval)
 	v.SetDefault("waiting-txs-limit", sidecar.DefaultWaitingTxsLimit)
 	v.SetDefault("channel-buffer-size", sidecar.DefaultBufferSize)

docs/architecture.md

@@ -93,7 +93,7 @@ The **Query Service** operates independently of the commit pipeline. Clients and
 
 The Sidecar acts as the critical bridge between the Hyperledger Fabric-X Ordering Service and the Committer's internal processing pipeline. It is responsible for fetching blocks sequentially from the Ordering Service, performing initial transaction validation to filter out malformed transactions, and maintaining a durable local block store on the file system.
 
-Beyond block ingestion, the Sidecar serves as the delivery endpoint for clients, providing committed blocks to registered applications and offering a notification service for transaction status updates. It also exposes query APIs that allow clients to fetch historical blocks and transactions directly from the block store.
+Beyond block ingestion, the Sidecar serves as the delivery endpoint for clients, providing committed blocks to registered applications and offering two notification mechanisms: transaction ID subscription for tracking specific transactions, and an all-transactions stream for monitoring all blockchain activity with optional filtering. It also exposes query APIs that allow clients to fetch historical blocks and transactions directly from the block store.
 
 **Key Characteristics:**
 

docs/notification-service.md

@@ -6,29 +6,42 @@ SPDX-License-Identifier: Apache-2.0
 
 # Notification Service — Client Usage Guide
 
-The Sidecar exposes a Notification Service that allows clients to subscribe to transaction status
-updates and receive asynchronous notifications when transactions are committed, rejected, or aborted.
-This is the primary mechanism for clients that submit transactions asynchronously to the Ordering
-Service to learn the outcome of their transactions, without polling or scanning the entire block stream.
+The Sidecar exposes a Notification Service that provides two mechanisms for receiving transaction status updates:
 
-The Notification Service uses a bidirectional gRPC stream: clients send subscription requests
-containing transaction IDs of interest, and the server pushes status responses as transactions
-complete. Multiple subscription requests can be sent on the same stream.
+1. **Transaction ID Subscription** — Allows clients to subscribe to transaction status
+   updates and receive asynchronous notifications when transactions are committed, rejected, or aborted.
+   This is the primary mechanism for clients that submit transactions asynchronously to the Ordering
+   Service to learn the outcome of their transactions, without polling or scanning the entire block stream.
+   It uses a bidirectional gRPC stream: clients send subscription requests
+   containing transaction IDs of interest, and the server pushes status responses as transactions
+   complete. Multiple subscription requests can be sent on the same stream.
+
+2. **All Transactions Stream** — Allows clients to subscribe to a stream of all committed transactions in block order,
+   with optional filtering by namespace and status. This is useful for audit, monitoring, event-driven applications, and
+   replication systems.
 
 For internal architecture details, see [sidecar.md — Section 6](sidecar.md#6-notification-service).
 
 ## 1. API Definition
 
-The Notification Service is defined as a bidirectional streaming RPC:
+The Notification Service provides two streaming RPCs:
 
 From [fabric-x-common/api/committerpb](https://github.com/hyperledger/fabric-x-common)
 
 ```protobuf
 service Notifier {
+    // Subscribe to specific transaction IDs
     rpc OpenNotificationStream (stream NotificationRequest) returns (stream NotificationResponse);
+
+    // Subscribe to all committed transactions
+    rpc StreamAllTransactions (StreamAllRequest) returns (stream TxBatch);
 }
 ```
 
+## 2. Transaction ID Subscription API
+
+### 2.1. API Definition
+
 The client sends `NotificationRequest` messages, each containing a batch of transaction IDs to watch
 and a timeout:
 
@@ -70,7 +83,7 @@ message RejectedTxIds {
 }
 ```
 
-## 2. Subscribing to Transaction Status Updates
+### 2.2. Subscribing to Transaction Status Updates
 
 Create a gRPC connection to the Sidecar, open a notification stream, and send a
 `NotificationRequest` with the transaction IDs to watch:
@@ -103,7 +116,7 @@ err = stream.Send(&committerpb.NotificationRequest{
 Multiple `NotificationRequest` messages can be sent on the same stream. Each request is
 tracked independently with its own timeout.
 
-## 3. Receiving Notifications
+## 2.3. Receiving Notifications
 
 The client receives `NotificationResponse` messages by calling `Recv()` on the stream.
 Each response contains one of the following payloads:
@@ -156,7 +169,7 @@ Responses are batched per stream for efficiency — if multiple subscribed trans
 complete in the same coordinator status update, they are grouped into a single
 `NotificationResponse`.
 
-## 4. Recommended Client Pattern
+### 2.4. Recommended Client Pattern
 
 To avoid missing notifications, clients should follow this sequence:
 
@@ -170,27 +183,119 @@ If the notification stream breaks (e.g., sidecar restart) or the timeout expires
 the transaction completes, the client should fall back to the Block Query API to check
 the transaction status.
 
-## 5. Concurrency Limits
+## 3. All Transactions Stream API
+
+### 3.1. API Definition
+
+The `StreamAllTransactions` RPC provides a server-streaming interface that delivers all committed transactions in block
+order. Unlike the transaction ID subscription API, this stream does not require clients to know transaction IDs in
+advance.
+
+```protobuf
+message StreamAllRequest {
+    repeated string filter_namespaces = 1;  // Optional: filter by namespace(s)
+    repeated Status filter_status = 2;      // Optional: filter by status
+    bool include_read_write_sets = 3;       // Optional: include read/write sets
+    bool include_endorsements = 4;          // Optional: include endorsements
+}
+
+message TxBatch {
+    repeated TxEvent events = 1;
+}
+
+message TxEvent {
+    TxRef ref = 1;                          // Transaction reference
+    Status status = 2;                      // Transaction status
+    repeated TxNamespace namespaces = 3;    // Namespaces (if filtering enabled)
+    repeated Endorsement endorsements = 4;  // Endorsements (if requested)
+}
+```
+
+### 3.2. Opening a Stream
+
+Create a gRPC connection to the Sidecar and call `StreamAllTransactions`:
+
+```go
+client := committerpb.NewNotifierClient(conn)
+stream, err := client.StreamAllTransactions(ctx, &committerpb.StreamAllRequest{
+    FilterNamespaces: []string{"namespace-1", "namespace-2"},
+    FilterStatus: []committerpb.Status{committerpb.Status_COMMITTED},
+    IncludeReadWriteSets: true,
+    IncludeEndorsements: false,
+})
+```
+
+**Request Parameters:**
+
+- **`filter_namespaces`** (optional): If specified, only transactions touching these namespaces are included. Empty
+  means all namespaces. Multiple namespaces use OR logic: a transaction is included if it touches any of the specified
+  namespaces.
+
+- **`filter_status`** (optional): If specified, only transactions with these status codes are included. Empty means all
+  statuses.
+
+- **`include_read_write_sets`** (optional): If true, the `namespaces` field in each `TxEvent`
+  includes the read/write sets for each namespace. Default: false.
+
+- **`include_endorsements`** (optional): If true, the `endorsements` field in each `TxEvent`
+  includes the transaction endorsements. Default: false.
+
+### 3.3. Receiving Transaction Events
+
+The server sends `TxBatch` messages containing one or more `TxEvent` entries. Transactions are batched by block. All
+transactions from the same block are delivered in a single `TxBatch`.
+
+```go
+for {
+    batch, err := stream.Recv()
+    ...
+
+    for _, event := range batch.Events {
+        ...
+    }
+}
+```
+
+### 3.4. Stream Behavior
+
+**Block Order Guarantee:**
+Transactions are streamed in deterministic block order. All transactions from block N are delivered before any
+transactions from block N+1.
+
+**Starting Point:**
+The stream starts from the currently processed block when the client connects. Historical transactions are not included.
+
+**No Recovery:**
+If the stream disconnects, the client must reconnect and will resume from the current block. Transactions from missed
+blocks are not replayed. For guaranteed delivery, use the Block Delivery API instead.
+
+**Backpressure:**
+If the client cannot keep up with the transaction rate, the server will block sending to that client. The server uses a
+configurable write timeout (default: 30 seconds) to prevent slow clients from blocking the system.
+
+## 4. Concurrency Limits
 
 The Notification Service shares the server's `max-concurrent-streams` limit (default: 10)
 with the Block Delivery streams ([Section 5 of sidecar.md](sidecar.md#5-block-delivery-service)).
-Both stream types compete for the same pool of stream slots.
+All stream types compete for the same pool of stream slots.
 
 When the limit is reached, new stream requests are rejected with a gRPC `ResourceExhausted`
 status code. Clients should handle this error with appropriate backoff and retry logic.
 
-## 6. Configuration
+## 5. Configuration
 
 The following configuration options in `sidecar.yaml` control notification behavior:
 
-| Setting | Default | Description |
-|---------|---------|-------------|
-| `notification.max-timeout` | `1m` | Upper limit on per-request timeout. Client timeouts are capped to this value. |
-| `server.max-concurrent-streams` | `10` | Maximum concurrent streaming RPCs across all stream types (Deliver + Notification). |
+| Setting                             | Default | Description                                                                                     |
+|-------------------------------------|---------|-------------------------------------------------------------------------------------------------|
+| `notification.max-timeout`          | `1m`    | Upper limit on per-request timeout for transaction ID subscriptions.                            |
+| `notification.stream-write-timeout` | `30s`   | Write timeout for all transactions stream. Prevents slow clients from blocking.                 |
+| `server.max-concurrent-streams`     | `10`    | Maximum concurrent streaming RPCs across all stream types (Deliver + Notification + StreamAll). |
 
 Sample configuration:
 
 ```yaml
 notification:
   max-timeout: 10m
+  stream-write-timeout: 10s
 ```

docs/sidecar.md

@@ -38,7 +38,9 @@ The Sidecar performs four main tasks:
 4. **Persist Committed Blocks:** Stores blocks confirmed as committed by the Coordinator in a local, append-only file
     store for durability and auditability.
 5. **Deliver to Clients:** Delivers the committed blocks to registered client applications.
-6. **Notification:** Notifies subscribers when transactions are committed or aborted, on a per-transaction ID basis.
+6. **Notification:** Provides two notification mechanisms:
+   - **Transaction ID Subscription:** Notifies subscribers when specific transactions (by ID) are committed or aborted.
+   - **All Transactions Stream:** Streams all committed transactions in block order with optional filtering.
 
 Note that the fourth task is executed only when users/clients creates a stream with the sidecar.
 
@@ -287,6 +289,8 @@ the status updates for all transactions within that block using `blockWithStatus
 ```go
 	blockWithStatus struct {
 		block         *common.Block
+		blockNumber   uint64
+		txs           []*protoblocktx.Tx
 		txStatus      []validationCode
 		txIDToTxIndex map[string]int
 		pendingCount  int
@@ -303,9 +307,14 @@ A block is only considered fully processed and ready for commitment when two con
 
 **g. Enqueueing Committed Blocks**: 
 
-When a block satisfies the criteria outlined in step f, the relay component first appends the collected
-transaction statuses within the metadata of the original Fabric block (sourced from `blocksToBeCommitted`). 
-Subsequently, this modified block is enqueued onto the `committedBlocks` output channel.
+When a block satisfies the criteria outlined in step f, the relay component performs two actions:
+
+1. Appends the collected transaction statuses within the metadata of the original Fabric block
+   (sourced from `blocksToBeCommitted`) and enqueues it onto the `committedBlocks` output channel
+   for persistence.
+
+2. Creates a `committedBlockWithTxs` structure containing the block number, transaction list, and
+   status information, and sends it to the notification service for distribution to subscribers.
 
 ### Task 3. Persisting Committed Block in the File System
 

go.mod

@@ -21,7 +21,7 @@ require (
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0
 	github.com/hyperledger/fabric-lib-go v1.1.3
 	github.com/hyperledger/fabric-protos-go-apiv2 v0.3.7
-	github.com/hyperledger/fabric-x-common v0.2.5-0.20260526151648-855f12b2d438
+	github.com/hyperledger/fabric-x-common v0.2.5
 	github.com/jackc/puddle/v2 v2.2.2
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/prometheus/client_golang v1.23.2

go.sum

@@ -188,8 +188,8 @@ github.com/hyperledger/fabric-lib-go v1.1.3 h1:alvgtBlbm373P3BLIxdHKVT4I64mORDwQ
 github.com/hyperledger/fabric-lib-go v1.1.3/go.mod h1:zwnGaLwmQ/usgC6Xbzh8jCnOniViRHSsg7WYErxbr30=
 github.com/hyperledger/fabric-protos-go-apiv2 v0.3.7 h1:sQ5qv8vQQfwewa1JlCiSCC8dLElmaU2/frLolpgibEY=
 github.com/hyperledger/fabric-protos-go-apiv2 v0.3.7/go.mod h1:bJnwzfv03oZQeCc863pdGTDgf5nmCy6Za3RAE7d2XsQ=
-github.com/hyperledger/fabric-x-common v0.2.5-0.20260526151648-855f12b2d438 h1:yVjqCihBd0YJ58XyldjlrGc+7+87fIftQE+eVXwrPlE=
-github.com/hyperledger/fabric-x-common v0.2.5-0.20260526151648-855f12b2d438/go.mod h1:EdyBG6jVFYYxJ5DgjxGVLE/Lav3GPhdftABZv5j+ttg=
+github.com/hyperledger/fabric-x-common v0.2.5 h1:AhyeIfAzLvLGIjf7RjDW5JPcnks2iNhpuek5rA1dmII=
+github.com/hyperledger/fabric-x-common v0.2.5/go.mod h1:EdyBG6jVFYYxJ5DgjxGVLE/Lav3GPhdftABZv5j+ttg=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/imkira/go-interpol v1.1.0 h1:KIiKr0VSG2CUW1hl1jpiyuzuJeKUUpC8iM1AIE7N1Vk=
 github.com/imkira/go-interpol v1.1.0/go.mod h1:z0h2/2T3XF8kyEPpRgJ3kmNv+C43p+I/CoI+jC3w2iA=

integration/runner/runtime.go

@@ -60,6 +60,7 @@ type (
 		SidecarClientConfig *connection.ClientConfig
 		NotifyClient        committerpb.NotifierClient
 		NotifyStream        committerpb.Notifier_OpenNotificationStreamClient
+		StreamAllTxStream   committerpb.Notifier_StreamAllTransactionsClient
 
 		CommittedBlock          chan *common.Block
 		TxBuilder               *workload.TxBuilder
@@ -303,6 +304,8 @@ func (c *CommitterRuntime) OpenNotificationStream(ctx context.Context, t *testin
 	var err error
 	c.NotifyStream, err = c.NotifyClient.OpenNotificationStream(ctx)
 	require.NoError(t, err)
+	c.StreamAllTxStream, err = c.NotifyClient.StreamAllTransactions(ctx, nil)
+	require.NoError(t, err)
 }
 
 // Start runs all services and load generator as configured by the serviceFlags.
@@ -549,7 +552,6 @@ func (c *CommitterRuntime) ValidateExpectedResultsInCommittedBlock(t *testing.T,
 	persistedTxIDsStatus := make([]*committerpb.TxStatus, 0, len(expected.TxIDs))
 	duplicateTxIDsStatus := make([]*committerpb.TxStatus, 0, len(expected.TxIDs))
 	for i, tID := range expected.TxIDs {
-		//nolint:gosec // int -> uint32.
 		s := committerpb.NewTxStatus(expected.Statuses[i], tID, blk.Header.Number, uint32(i))
 		if s.Status == committerpb.Status_REJECTED_DUPLICATE_TX_ID {
 			duplicateTxIDsStatus = append(duplicateTxIDsStatus, s)
@@ -575,6 +577,7 @@ func (c *CommitterRuntime) ValidateExpectedResultsInCommittedBlock(t *testing.T,
 	}
 
 	sidecar.RequireNotifications(t, c.NotifyStream, blk.Header.Number, expected.TxIDs, expected.Statuses)
+	sidecar.RequireStreamAllTransactions(t, c.StreamAllTxStream, blk.Header.Number, expected.TxIDs, expected.Statuses)
 }
 
 // CountStatus returns the number of transactions with a given tx status.