Tolerate missing ns.coll in change event. (#149)

Missing `ns.coll` can happen if, e.g., there’s a `dropDatabase` event. This changeset tolerates that. (We don’t support such events on the source, but on the destination we ignore them.)

The fix rewrites the UnmarshalBSON method to avoid LookupErr. Pprof shows it slightly more efficient to do things this way, and it better mimics the driver’s own tolerance missing fields when unmarshaling.

Author: FGasper

internal/verifier/util.go

@@ -67,15 +67,29 @@ func (ns *Namespace) FullName() string {
 // avoid reflection, which can substantially impede performance in “hot”
 // code paths like this.
 func (ns *Namespace) UnmarshalBSON(in []byte) error {
-	rawDoc := bson.Raw(in)
-
-	err := mbson.LookupTo(rawDoc, &ns.DB, "db")
-
-	if err != nil {
-		return err
+	for el, err := range mbson.RawElements(in) {
+		if err != nil {
+			return errors.Wrap(err, "iterating BSON fields")
+		}
+
+		key, err := el.KeyErr()
+		if err != nil {
+			return errors.Wrap(err, "reading BSON field name")
+		}
+
+		switch key {
+		case "db":
+			if err := mbson.UnmarshalElementValue(el, &ns.DB); err != nil {
+				return err
+			}
+		case "coll":
+			if err := mbson.UnmarshalElementValue(el, &ns.Coll); err != nil {
+				return err
+			}
+		}
 	}
 
-	return mbson.LookupTo(rawDoc, &ns.Coll, "coll")
+	return nil
 }
 
 // NewNamespace returns a new Namespace struct with the given parameters.

mbson/bson_raw.go

@@ -1,6 +1,9 @@
 package mbson
 
 import (
+	"fmt"
+	"iter"
+
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
@@ -28,6 +31,41 @@ func RawContains(doc bson.Raw, keys ...string) (bool, error) {
 	return RawLookup(doc, &val, keys...)
 }
 
+// RawElements returns an iterator over a Raw’s elements.
+func RawElements(doc bson.Raw) iter.Seq2[bson.RawElement, error] {
+	remaining := doc[4:]
+
+	return func(yield func(bson.RawElement, error) bool) {
+		var el bsoncore.Element
+		var ok bool
+
+		for len(remaining) > 1 {
+			el, remaining, ok = bsoncore.ReadElement(remaining)
+
+			var err error
+
+			if !ok {
+				err = bsoncore.NewInsufficientBytesError(doc, remaining)
+			} else {
+				err = el.Validate()
+			}
+
+			if err != nil {
+				if yield(nil, err) {
+					panic(fmt.Sprintf("Must stop iteration after error (%v)", err))
+				}
+
+				return
+			}
+
+			if !yield(bson.RawElement(el), nil) {
+				return
+			}
+		}
+
+	}
+}
+
 // ConvertToRawValue converts the specified argument to a bson.RawValue.
 func ConvertToRawValue(thing any) (bson.RawValue, error) {
 	if thing == nil {

mbson/raw_test.go

@@ -0,0 +1,62 @@
+package mbson
+
+import (
+	"testing"
+
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+func TestRawElements(t *testing.T) {
+	srcD := bson.D{
+		{"foo", "xxx"},
+		{"bar", "baz"},
+	}
+
+	mydoc := lo.Must(bson.Marshal(srcD))
+
+	received := bson.D{}
+
+	for el, err := range RawElements(mydoc) {
+		require.NoError(t, err, "should iterate")
+
+		received = append(received, bson.E{
+			lo.Must(el.KeyErr()),
+			lo.Must(el.ValueErr()).StringValue(),
+		})
+	}
+
+	assert.Equal(t, srcD, received, "should iterate all fields")
+
+	// Now make the document invalid
+	mydoc = mydoc[:len(mydoc)-3]
+
+	received = received[:0]
+
+	var iterErr error
+	for _, err := range RawElements(mydoc) {
+		if err != nil {
+			iterErr = err
+			break
+		}
+	}
+
+	assert.Error(t, iterErr, "should fail somewhere in the iteration")
+
+	assert.Panics(
+		t,
+		func() {
+			var gotErr error
+			for _, err := range RawElements(mydoc) {
+				if err != nil {
+					gotErr = err
+				}
+			}
+
+			assert.Error(t, gotErr, "should get an error")
+		},
+		"should panic if we fail but didn’t stop iterating",
+	)
+}