fix: Filter subscription events by collection before opening txn

client/request/select.go

@@ -39,6 +39,20 @@ type Select struct {
 	// IsEncrypted indicates that this is an encrypted query that should
 	// use searchable encryption to query remote nodes.
 	IsEncrypted bool
+
+	// targetCollectionID is the resolved root CollectionID for this select's
+	// target collection. It is set by the subscription event loop via
+	// SetTargetCollectionID at subscribe-time and consulted via
+	// CheckCollectionFilter to drop events from other collections without
+	// opening a transaction. Outside the subscription path this remains "".
+	targetCollectionID string
+}
+
+// SetTargetCollectionID records the resolved root CollectionID this select
+// targets, so subsequent CheckCollectionFilter calls can reject events from
+// other collections. Intended only for the subscription handler.
+func (s *Select) SetTargetCollectionID(id string) {
+	s.targetCollectionID = id
 }
 
 // ChildSelect represents a type with selectable child properties.
@@ -147,6 +161,14 @@ func (s *Select) CheckCIDFilter(cid string) bool {
 	return !s.CIDs.HasValue() || slices.Contains(s.CIDs.Value(), cid)
 }
 
+// CheckCollectionFilter checks if the given root CollectionID matches the
+// select's resolved target collection. Returns true if the IDs match, or if
+// no target has been recorded (preserving current behaviour for callers
+// outside the subscription path).
+func (s *Select) CheckCollectionFilter(collectionID string) bool {
+	return s.targetCollectionID == "" || s.targetCollectionID == collectionID
+}
+
 // CheckDocIDFilter checks if the given docID passes the DocID filter.
 // Returns true if the docID passes the filter, false otherwise.
 // If no DocID filter is set, it always passes.

internal/db/subscription_collection_filter_test.go

@@ -0,0 +1,134 @@
+// Copyright 2026 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package db
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/sourcenetwork/defradb/client"
+)
+
+const bookAuthorSchema = `
+type Book {
+	title: String
+	author: Author
+	publisher: Publisher
+}
+type Author {
+	name: String
+	wrote: Book @primary
+}
+type Publisher {
+	label: String
+	published: Book @primary
+}
+`
+
+// A Book subscription must NOT open a new transaction in response to an
+// Author event. db.previousTxnID is an atomic counter incremented inside
+// every db.NewTxn call (db.go:226-232); reading it lets us observe whether
+// the subscription event loop reached its db.NewTxn call (subscriptions.go:74)
+// for the wrong-collection event.
+//
+// Pre-fix:  counter advances by >=1 — the subscription opens its own txn
+//
+//	before the planner fails inside VersionedFetcher.merge().
+//
+// Post-fix: counter is unchanged — CheckCollectionFilter rejected the event
+//
+//	at the docID/CID filter site (subscriptions.go:71) before any
+//	transaction was opened.
+func TestHandleSubscription_WrongCollectionEvent_OpensNoTxn(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	db, err := newBadgerDB(ctx)
+	require.NoError(t, err)
+	defer db.Close()
+
+	_, err = db.AddCollection(ctx, bookAuthorSchema)
+	require.NoError(t, err)
+
+	authorCol, err := db.GetCollectionByName(ctx, "Author")
+	require.NoError(t, err)
+
+	res := db.ExecRequest(
+		ctx,
+		`subscription { Book { _docID title author { name } publisher { label } } }`,
+	)
+	require.Empty(t, res.GQL.Errors)
+	subCh := res.Subscription
+	require.NotNil(t, subCh)
+
+	authorDoc, err := client.NewDocFromJSON(ctx, []byte(`{"name": "Tolkien"}`), authorCol.Version())
+	require.NoError(t, err)
+	require.NoError(t, authorCol.AddDocument(ctx, authorDoc))
+
+	// Capture the txn counter AFTER the mutation's own work returned.
+	// The subscription event handler runs on its own goroutine; we then
+	// wait and assert the counter has not advanced further.
+	mid := db.previousTxnID.Load()
+	time.Sleep(200 * time.Millisecond)
+	after := db.previousTxnID.Load()
+
+	require.Equal(t, mid, after,
+		"subscription must not open a transaction for a wrong-collection event; counter advanced from %d to %d",
+		mid, after)
+
+	// Belt-and-braces: also confirm nothing surfaces on the response channel.
+	select {
+	case got, ok := <-subCh:
+		if !ok {
+			t.Fatalf("subscription channel closed unexpectedly")
+		}
+		t.Fatalf("expected no response for wrong-collection event, got %+v", got)
+	default:
+	}
+}
+
+// Control: same-collection events must still be delivered. Guards against the
+// new collection filter being over-aggressive.
+func TestHandleSubscription_RightCollectionEvent_StillDelivered(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	db, err := newBadgerDB(ctx)
+	require.NoError(t, err)
+	defer db.Close()
+
+	_, err = db.AddCollection(ctx, bookAuthorSchema)
+	require.NoError(t, err)
+
+	bookCol, err := db.GetCollectionByName(ctx, "Book")
+	require.NoError(t, err)
+
+	res := db.ExecRequest(ctx, `subscription { Book { _docID title } }`)
+	require.Empty(t, res.GQL.Errors)
+	subCh := res.Subscription
+	require.NotNil(t, subCh)
+
+	bookDoc, err := client.NewDocFromJSON(ctx, []byte(`{"title": "The Hobbit"}`), bookCol.Version())
+	require.NoError(t, err)
+	require.NoError(t, bookCol.AddDocument(ctx, bookDoc))
+
+	select {
+	case got, ok := <-subCh:
+		require.True(t, ok, "subscription channel must stay open")
+		require.Empty(t, got.Errors, "right-collection event must deliver without errors")
+		require.NotEmpty(t, got.Data)
+	case <-time.After(500 * time.Millisecond):
+		t.Fatalf("expected right-collection event to be delivered")
+	}
+}

internal/db/subscriptions.go

@@ -25,6 +25,15 @@ type subscriptionSelector interface {
 	ToSubscriptionSelect(docID, cid string) request.Selection
 	CheckCIDFilter(cid string) bool
 	CheckDocIDFilter(docID string) bool
+	CheckCollectionFilter(collectionID string) bool
+}
+
+// targetCollectionSetter is implemented by subscription selectors whose
+// target collection can be resolved at subscribe-time. The subscription
+// handler uses this to inject the root CollectionID so CheckCollectionFilter
+// can drop events from other collections without opening a transaction.
+type targetCollectionSetter interface {
+	SetTargetCollectionID(id string)
 }
 
 // handleSubscription checks for a subscription within the given request and
@@ -39,6 +48,22 @@ func (db *DB) handleSubscription(ctx context.Context, r *request.Request) (<-cha
 		return nil, client.NewErrUnexpectedType[request.Selection]("SubscriptionSelection", subRequest)
 	}
 
+	// Resolve the target collection's root CollectionID once at subscribe-time
+	// so the event loop can reject events from other collections without
+	// opening a transaction. If the target isn't a request.Select we skip the
+	// stamping; CheckCollectionFilter will default to "allow" for such cases.
+	if setter, ok := subRequest.(targetCollectionSetter); ok {
+		selection, ok := r.Subscription[0].Selections[0].(*request.Select)
+		if !ok {
+			return nil, client.NewErrUnexpectedType[request.Selection]("SubscriptionSelection", subRequest)
+		}
+		col, err := db.GetCollectionByName(ctx, selection.Name)
+		if err != nil {
+			return nil, err
+		}
+		setter.SetTargetCollectionID(col.Version().CollectionID)
+	}
+
 	sub, err := db.events.Subscribe(event.UpdateName)
 	if err != nil {
 		return nil, err
@@ -65,10 +90,13 @@ func (db *DB) handleSubscription(ctx context.Context, r *request.Request) (<-cha
 					continue // invalid event value
 				}
 			}
-			// Skip events that do not pass the subscription's docID and cid filters
-			// This is an optimization to avoid running the selection planner and
-			// related query logic when we know the event will not be relevant to the subscription.
-			if !subRequest.CheckDocIDFilter(evt.DocID) || !subRequest.CheckCIDFilter(evt.Cid.String()) {
+			// Skip events that do not pass the subscription's collection, docID,
+			// and cid filters. This is an optimization to avoid running the
+			// selection planner and related query logic when we know the event
+			// will not be relevant to the subscription.
+			if !subRequest.CheckCollectionFilter(evt.CollectionID) ||
+				!subRequest.CheckDocIDFilter(evt.DocID) ||
+				!subRequest.CheckCIDFilter(evt.Cid.String()) {
 				continue
 			}
 			txn, err := db.NewTxn(false)