diff --git a/connectors/mongo/conn.go b/connectors/mongo/conn.go
index aa74c47b..c403bac9 100644
--- a/connectors/mongo/conn.go
+++ b/connectors/mongo/conn.go
@@ -896,45 +896,166 @@ func (c *conn) WriteData(ctx context.Context, r *connect.Request[adiomv1.WriteDa
 	return connect.NewResponse(&adiomv1.WriteDataResponse{}), nil
 }
 
-// WriteUpdates implements adiomv1connect.ConnectorServiceHandler.
-func (c *conn) WriteUpdates(ctx context.Context, r *connect.Request[adiomv1.WriteUpdatesRequest]) (*connect.Response[adiomv1.WriteUpdatesResponse], error) {
-	col, _, ok := GetCol(c.client, r.Msg.GetNamespace())
-	if !ok {
-		return nil, connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("namespace should be fully qualified"))
+func (c *conn) buildIdFilter(update *adiomv1.Update) (bson.D, string, error) {
+	var idFilter bson.D
+	var normalized []*adiomv1.BsonValue
+	if len(update.GetId()) == 0 {
+		return nil, "", fmt.Errorf("err update with unexpected empty id")
 	}
-	updates := util.KeepLastUpdate(r.Msg.GetUpdates())
-	var models []mongo.WriteModel
-	for _, update := range updates {
-		var idFilter bson.D
-		if len(update.GetId()) == 0 {
-			return nil, connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("err update with unexpected empty id"))
+	for _, idPart := range update.GetId() {
+		key := idPart.GetName()
+		// For backwards compatibility- we used to pass no key
+		if key == "" {
+			key = "_id"
 		}
-		for _, idPart := range update.GetId() {
-			key := idPart.GetName()
-			// For backwards compatibility- we used to pass no key
-			if key == "" {
-				key = "_id"
-			}
-			if !c.settings.FullDocumentKey && key != "_id" {
-				continue
-			}
-			typ := bson.Type(idPart.GetType())
-			idFilter = append(idFilter, bson.E{Key: key, Value: bson.RawValue{Type: typ, Value: idPart.GetData()}})
+		if !c.settings.FullDocumentKey && key != "_id" {
+			continue
+		}
+		typ := bson.Type(idPart.GetType())
+		idFilter = append(idFilter, bson.E{Key: key, Value: bson.RawValue{Type: typ, Value: idPart.GetData()}})
+		normalized = append(normalized, &adiomv1.BsonValue{Name: key, Type: idPart.GetType(), Data: idPart.GetData()})
+	}
+	if len(idFilter) == 0 {
+		return nil, "", fmt.Errorf("err with _id not found- enable full-document-key or ensure an id part has the name _id: %v", update.GetId())
+	}
+	return idFilter, util.BsonIdKey(normalized), nil
+}
+
+func stripIdFields(raw bson.Raw, idFilter bson.D) (bson.Raw, error) {
+	idKeys := make(map[string]struct{}, len(idFilter))
+	for _, e := range idFilter {
+		idKeys[e.Key] = struct{}{}
+	}
+	elems, err := raw.Elements()
+	if err != nil {
+		return nil, fmt.Errorf("stripIdFields: malformed bson: %w", err)
+	}
+	filtered := make(bson.D, 0, len(elems))
+	stripped := false
+	for _, elem := range elems {
+		if _, ok := idKeys[elem.Key()]; ok {
+			stripped = true
+			continue
+		}
+		filtered = append(filtered, bson.E{Key: elem.Key(), Value: elem.Value()})
+	}
+	if !stripped {
+		return raw, nil
+	}
+	if len(filtered) == 0 {
+		return nil, nil
+	}
+	out, err := bson.Marshal(filtered)
+	if err != nil {
+		return nil, fmt.Errorf("stripIdFields: marshal filtered doc: %w", err)
+	}
+	return out, nil
+}
+
+// buildBulkModels converts a sequence of updates into a slice of mongo.WriteModel
+// ready for BulkWrite, along with whether the batch must run ordered.
+//
+// Iterates updates in reverse so that per-ID dedup keeps the last (chronological) operation.
+// When the last operation for an ID is a partial update, earlier operations for the same ID
+// are kept as ordered models so they execute in sequence before the unordered batch.
+func (c *conn) buildBulkModels(updates []*adiomv1.Update) ([]mongo.WriteModel, bool, error) {
+	var models []mongo.WriteModel
+	var orderedModels []mongo.WriteModel
+
+	seen := map[string]adiomv1.UpdateType{}
+
+	for i := len(updates) - 1; i >= 0; i-- {
+		update := updates[i]
+		idFilter, idKey, err := c.buildIdFilter(update)
+		if err != nil {
+			return nil, false, err
 		}
-		if len(idFilter) == 0 {
-			return nil, connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("err with _id not found- enable full-document-key or ensure an id part has the name _id: %v", update.GetId()))
+
+		lastType, found := seen[idKey]
+		isNew := !found
+		if isNew {
+			seen[idKey] = update.GetType()
+			lastType = update.GetType()
 		}
+
 		switch update.GetType() {
 		case adiomv1.UpdateType_UPDATE_TYPE_INSERT, adiomv1.UpdateType_UPDATE_TYPE_UPDATE:
 			model := mongo.NewReplaceOneModel().SetFilter(idFilter).SetReplacement(bson.Raw(update.GetData())).SetUpsert(true)
-			models = append(models, model)
+			if isNew {
+				models = append(models, model)
+			} else if lastType == adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE {
+				orderedModels = append(orderedModels, model)
+			}
 		case adiomv1.UpdateType_UPDATE_TYPE_DELETE:
 			model := mongo.NewDeleteOneModel().SetFilter(idFilter)
-			models = append(models, model)
+			if isNew {
+				models = append(models, model)
+			} else if lastType == adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE {
+				orderedModels = append(orderedModels, model)
+			}
+		case adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE:
+			theUpdate := bson.M{}
+			if update.GetData() != nil {
+				stripped, err := stripIdFields(bson.Raw(update.GetData()), idFilter)
+				if err != nil {
+					return nil, false, err
+				}
+				if len(stripped) > 0 {
+					theUpdate["$set"] = stripped
+				}
+			}
+			if len(update.GetPartialUpdateUnset()) > 0 {
+				idKeys := make(map[string]struct{}, len(idFilter))
+				for _, e := range idFilter {
+					idKeys[e.Key] = struct{}{}
+				}
+				inner := bson.M{}
+				for _, field := range update.GetPartialUpdateUnset() {
+					if _, isId := idKeys[field]; isId {
+						continue
+					}
+					inner[field] = 1
+				}
+				if len(inner) > 0 {
+					theUpdate["$unset"] = inner
+				}
+			}
+			if len(theUpdate) == 0 {
+				continue
+			}
+			model := mongo.NewUpdateOneModel().SetFilter(idFilter).SetUpdate(theUpdate).SetUpsert(true)
+			if isNew {
+				models = append(models, model)
+			} else if lastType == adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE {
+				orderedModels = append(orderedModels, model)
+			}
 		}
 	}
 
-	_, err := col.BulkWrite(ctx, models, options.BulkWrite().SetOrdered(false))
+	if len(orderedModels) > 0 {
+		slices.Reverse(orderedModels)
+		return append(orderedModels, models...), true, nil
+	}
+	return models, false, nil
+}
+
+// WriteUpdates implements adiomv1connect.ConnectorServiceHandler.
+func (c *conn) WriteUpdates(ctx context.Context, r *connect.Request[adiomv1.WriteUpdatesRequest]) (*connect.Response[adiomv1.WriteUpdatesResponse], error) {
+	col, _, ok := GetCol(c.client, r.Msg.GetNamespace())
+	if !ok {
+		return nil, connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("namespace should be fully qualified"))
+	}
+
+	finalModels, ordered, err := c.buildBulkModels(r.Msg.GetUpdates())
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInvalidArgument, err)
+	}
+
+	if len(finalModels) == 0 {
+		return connect.NewResponse(&adiomv1.WriteUpdatesResponse{}), nil
+	}
+
+	_, err = col.BulkWrite(ctx, finalModels, options.BulkWrite().SetOrdered(ordered))
 	if err != nil {
 		if !errors.Is(err, context.Canceled) {
 			slog.Error(fmt.Sprintf("Failed to insert bulk updates: %v", err))
diff --git a/connectors/mongo/conn_unit_test.go b/connectors/mongo/conn_unit_test.go
new file mode 100644
index 00000000..82ed5739
--- /dev/null
+++ b/connectors/mongo/conn_unit_test.go
@@ -0,0 +1,352 @@
+/*
+ * Copyright (C) 2024 Adiom, Inc.
+ *
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+package mongo
+
+import (
+	"testing"
+
+	adiomv1 "github.com/adiom-data/dsync/gen/adiom/v1"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+)
+
+// helpers
+
+func mustMarshal(t *testing.T, v interface{}) []byte {
+	t.Helper()
+	b, err := bson.Marshal(v)
+	require.NoError(t, err)
+	return b
+}
+
+func bsonID(t *testing.T, name string, v interface{}) *adiomv1.BsonValue {
+	t.Helper()
+	if name == "" {
+		name = "_id"
+	}
+	typ, data, err := bson.MarshalValue(v)
+	require.NoError(t, err)
+	return &adiomv1.BsonValue{Name: name, Type: uint32(typ), Data: data}
+}
+
+// buildIdFilter
+
+func TestBuildIdFilter_EmptyId(t *testing.T) {
+	c := &conn{}
+	_, _, err := c.buildIdFilter(&adiomv1.Update{})
+	assert.ErrorContains(t, err, "unexpected empty id")
+}
+
+func TestBuildIdFilter_BackwardsCompatNoName(t *testing.T) {
+	c := &conn{}
+	// A single id part without a name should be treated as _id.
+	part := bsonID(t, "", "abc")
+	part.Name = ""
+	filter, key, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{part}})
+	require.NoError(t, err)
+	require.Len(t, filter, 1)
+	assert.Equal(t, "_id", filter[0].Key)
+	assert.NotEmpty(t, key)
+}
+
+func TestBuildIdFilter_NoIdWhenFullDocKeyDisabled(t *testing.T) {
+	c := &conn{settings: ConnectorSettings{FullDocumentKey: false}}
+	// Non-_id parts get filtered, leaving empty idFilter.
+	_, _, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{bsonID(t, "shard", "s1")}})
+	assert.ErrorContains(t, err, "_id not found")
+}
+
+func TestBuildIdFilter_FullDocumentKeyKeepsAllParts(t *testing.T) {
+	c := &conn{settings: ConnectorSettings{FullDocumentKey: true}}
+	filter, key, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{
+		bsonID(t, "_id", "doc1"),
+		bsonID(t, "shard", "s1"),
+	}})
+	require.NoError(t, err)
+	require.Len(t, filter, 2)
+	assert.Equal(t, "_id", filter[0].Key)
+	assert.Equal(t, "shard", filter[1].Key)
+	assert.NotEmpty(t, key)
+}
+
+func TestBuildIdFilter_DedupKeyDiffersByNormalizedId(t *testing.T) {
+	c := &conn{settings: ConnectorSettings{FullDocumentKey: false}}
+	_, k1, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{
+		bsonID(t, "_id", "doc1"),
+		bsonID(t, "shard", "s1"),
+	}})
+	require.NoError(t, err)
+	_, k2, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{
+		bsonID(t, "_id", "doc1"),
+		bsonID(t, "shard", "s2"),
+	}})
+	require.NoError(t, err)
+	// FullDocumentKey=false -> both collapse to just _id=doc1, so keys match.
+	assert.Equal(t, k1, k2)
+
+	c.settings.FullDocumentKey = true
+	_, k3, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{
+		bsonID(t, "_id", "doc1"),
+		bsonID(t, "shard", "s1"),
+	}})
+	require.NoError(t, err)
+	_, k4, err := c.buildIdFilter(&adiomv1.Update{Id: []*adiomv1.BsonValue{
+		bsonID(t, "_id", "doc1"),
+		bsonID(t, "shard", "s2"),
+	}})
+	require.NoError(t, err)
+	// With FullDocumentKey=true the shard value must be part of the key.
+	assert.NotEqual(t, k3, k4)
+}
+
+// stripIdFields
+
+func TestStripIdFields_NoStripNeeded(t *testing.T) {
+	raw := bson.Raw(mustMarshal(t, bson.M{"a": 1, "b": 2}))
+	idFilter := bson.D{{Key: "_id", Value: "x"}}
+	out, err := stripIdFields(raw, idFilter)
+	require.NoError(t, err)
+	// Bytes are unchanged when no id fields are present.
+	assert.Equal(t, []byte(raw), []byte(out))
+}
+
+func TestStripIdFields_StripsIdField(t *testing.T) {
+	raw := bson.Raw(mustMarshal(t, bson.M{"_id": "x", "a": 1}))
+	idFilter := bson.D{{Key: "_id", Value: "x"}}
+	out, err := stripIdFields(raw, idFilter)
+	require.NoError(t, err)
+	var decoded bson.M
+	require.NoError(t, bson.Unmarshal(out, &decoded))
+	assert.NotContains(t, decoded, "_id")
+	assert.Equal(t, int32(1), decoded["a"])
+}
+
+func TestStripIdFields_OnlyIdReturnsNil(t *testing.T) {
+	raw := bson.Raw(mustMarshal(t, bson.M{"_id": "x"}))
+	idFilter := bson.D{{Key: "_id", Value: "x"}}
+	out, err := stripIdFields(raw, idFilter)
+	require.NoError(t, err)
+	assert.Nil(t, out)
+}
+
+func TestStripIdFields_StripsCompositeIdFields(t *testing.T) {
+	raw := bson.Raw(mustMarshal(t, bson.M{"_id": "x", "shard": "s", "a": 1}))
+	idFilter := bson.D{
+		{Key: "_id", Value: "x"},
+		{Key: "shard", Value: "s"},
+	}
+	out, err := stripIdFields(raw, idFilter)
+	require.NoError(t, err)
+	var decoded bson.M
+	require.NoError(t, bson.Unmarshal(out, &decoded))
+	assert.NotContains(t, decoded, "_id")
+	assert.NotContains(t, decoded, "shard")
+	assert.Equal(t, int32(1), decoded["a"])
+}
+
+func TestStripIdFields_MalformedBsonReturnsError(t *testing.T) {
+	_, err := stripIdFields(bson.Raw([]byte{0x01, 0x02, 0x03}), bson.D{{Key: "_id", Value: "x"}})
+	assert.Error(t, err)
+}
+
+// buildBulkModels
+
+func insertU(t *testing.T, id string, data bson.M) *adiomv1.Update {
+	return &adiomv1.Update{
+		Id:   []*adiomv1.BsonValue{bsonID(t, "_id", id)},
+		Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+		Data: mustMarshal(t, data),
+	}
+}
+
+func deleteU(t *testing.T, id string) *adiomv1.Update {
+	return &adiomv1.Update{
+		Id:   []*adiomv1.BsonValue{bsonID(t, "_id", id)},
+		Type: adiomv1.UpdateType_UPDATE_TYPE_DELETE,
+	}
+}
+
+func partialU(t *testing.T, id string, data bson.M, unset ...string) *adiomv1.Update {
+	u := &adiomv1.Update{
+		Id:                  []*adiomv1.BsonValue{bsonID(t, "_id", id)},
+		Type:                adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE,
+		PartialUpdateUnset:  unset,
+	}
+	if data != nil {
+		u.Data = mustMarshal(t, data)
+	}
+	return u
+}
+
+func TestBuildBulkModels_Empty(t *testing.T) {
+	c := &conn{}
+	models, ordered, err := c.buildBulkModels(nil)
+	require.NoError(t, err)
+	assert.Empty(t, models)
+	assert.False(t, ordered)
+}
+
+func TestBuildBulkModels_AllUniqueStaysUnordered(t *testing.T) {
+	c := &conn{}
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{
+		insertU(t, "a", bson.M{"x": 1}),
+		insertU(t, "b", bson.M{"x": 2}),
+		deleteU(t, "c"),
+	})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	assert.Len(t, models, 3)
+}
+
+func TestBuildBulkModels_DedupKeepsLast(t *testing.T) {
+	c := &conn{}
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{
+		insertU(t, "a", bson.M{"v": 1}),
+		insertU(t, "a", bson.M{"v": 2}),
+		insertU(t, "a", bson.M{"v": 3}),
+	})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	require.Len(t, models, 1)
+	replace, ok := models[0].(*mongo.ReplaceOneModel)
+	require.True(t, ok)
+	var doc bson.M
+	require.NoError(t, bson.Unmarshal(replace.Replacement.(bson.Raw), &doc))
+	assert.Equal(t, int32(3), doc["v"])
+}
+
+func TestBuildBulkModels_DeleteAfterInsertsKeepsOnlyDelete(t *testing.T) {
+	c := &conn{}
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{
+		insertU(t, "a", bson.M{"v": 1}),
+		insertU(t, "a", bson.M{"v": 2}),
+		deleteU(t, "a"),
+	})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	require.Len(t, models, 1)
+	_, ok := models[0].(*mongo.DeleteOneModel)
+	assert.True(t, ok)
+}
+
+func TestBuildBulkModels_PartialLastTriggersOrdered(t *testing.T) {
+	c := &conn{}
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{
+		insertU(t, "a", bson.M{"v": 1}),
+		partialU(t, "a", bson.M{"w": 2}),
+	})
+	require.NoError(t, err)
+	assert.True(t, ordered)
+	require.Len(t, models, 2)
+	// Ordered prefix runs first: the original insert, then the partial.
+	_, isReplace := models[0].(*mongo.ReplaceOneModel)
+	assert.True(t, isReplace)
+	_, isUpdate := models[1].(*mongo.UpdateOneModel)
+	assert.True(t, isUpdate)
+}
+
+func TestBuildBulkModels_PartialLastOrderedPreservesChronology(t *testing.T) {
+	c := &conn{}
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{
+		insertU(t, "a", bson.M{"v": 1}),
+		partialU(t, "a", bson.M{"w": 2}),
+		insertU(t, "a", bson.M{"v": 3}),
+		partialU(t, "a", bson.M{"x": 4}),
+	})
+	require.NoError(t, err)
+	assert.True(t, ordered)
+	// Four ops, all for the same id; ordered must preserve chronological order.
+	require.Len(t, models, 4)
+	_, m0 := models[0].(*mongo.ReplaceOneModel)
+	_, m1 := models[1].(*mongo.UpdateOneModel)
+	_, m2 := models[2].(*mongo.ReplaceOneModel)
+	_, m3 := models[3].(*mongo.UpdateOneModel)
+	assert.True(t, m0 && m1 && m2 && m3)
+}
+
+func TestBuildBulkModels_PartialOnlyIdInDataSkipped(t *testing.T) {
+	c := &conn{}
+	// Partial with data that contains only _id (stripped to nil) and no unset -> no-op.
+	u := partialU(t, "a", bson.M{"_id": "a"})
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{u})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	assert.Empty(t, models)
+}
+
+func TestBuildBulkModels_PartialUnsetOnly(t *testing.T) {
+	c := &conn{}
+	u := partialU(t, "a", nil, "foo", "bar")
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{u})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	require.Len(t, models, 1)
+	upd, ok := models[0].(*mongo.UpdateOneModel)
+	require.True(t, ok)
+	body := upd.Update.(bson.M)
+	assert.NotContains(t, body, "$set")
+	unset, ok := body["$unset"].(bson.M)
+	require.True(t, ok)
+	assert.Contains(t, unset, "foo")
+	assert.Contains(t, unset, "bar")
+}
+
+func TestBuildBulkModels_PartialUnsetFiltersIdFields(t *testing.T) {
+	c := &conn{}
+	u := partialU(t, "a", bson.M{"x": 1}, "_id", "foo")
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{u})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	require.Len(t, models, 1)
+	upd, ok := models[0].(*mongo.UpdateOneModel)
+	require.True(t, ok)
+	body := upd.Update.(bson.M)
+	unset, ok := body["$unset"].(bson.M)
+	require.True(t, ok)
+	assert.NotContains(t, unset, "_id")
+	assert.Contains(t, unset, "foo")
+}
+
+func TestBuildBulkModels_PartialUnsetOnlyIdBecomesNoop(t *testing.T) {
+	c := &conn{}
+	// Unset only contains _id which gets filtered -> empty update -> skipped.
+	u := partialU(t, "a", nil, "_id")
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{u})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	assert.Empty(t, models)
+}
+
+func TestBuildBulkModels_FullDocumentKeyDedupesByComposite(t *testing.T) {
+	c := &conn{settings: ConnectorSettings{FullDocumentKey: true}}
+	mk := func(id, shard string, v int) *adiomv1.Update {
+		return &adiomv1.Update{
+			Id: []*adiomv1.BsonValue{
+				bsonID(t, "_id", id),
+				bsonID(t, "shard", shard),
+			},
+			Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+			Data: mustMarshal(t, bson.M{"v": v}),
+		}
+	}
+	models, ordered, err := c.buildBulkModels([]*adiomv1.Update{
+		mk("a", "s1", 1),
+		mk("a", "s2", 2), // same _id, different shard -> distinct doc
+		mk("a", "s1", 3), // same composite as first -> dedup replaces it
+	})
+	require.NoError(t, err)
+	assert.False(t, ordered)
+	// Two unique docs after dedup.
+	assert.Len(t, models, 2)
+}
+
+func TestBuildBulkModels_ErrorPropagates(t *testing.T) {
+	c := &conn{}
+	_, _, err := c.buildBulkModels([]*adiomv1.Update{{Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT}})
+	assert.ErrorContains(t, err, "unexpected empty id")
+}
diff --git a/connectors/mongo/connector_test.go b/connectors/mongo/connector_test.go
index d1f7aa77..379c66b6 100644
--- a/connectors/mongo/connector_test.go
+++ b/connectors/mongo/connector_test.go
@@ -15,6 +15,7 @@ import (
 	"testing"
 	"time"
 
+	"connectrpc.com/connect"
 	"github.com/adiom-data/dsync/connectors/common"
 	adiomv1 "github.com/adiom-data/dsync/gen/adiom/v1"
 	"github.com/adiom-data/dsync/gen/adiom/v1/adiomv1connect"
@@ -64,6 +65,135 @@ func TestMongoConnectorSuite(t *testing.T) {
 	suite.Run(t, tSuite)
 }
 
+func assertDoc(t *testing.T, col *mongo.Collection, expected map[string]string) {
+	var res map[string]string
+	col.FindOne(t.Context(), bson.M{"_id": expected["_id"]}).Decode(&res)
+	assert.Equal(t, expected, res)
+}
+
+func assertNoDoc(t *testing.T, col *mongo.Collection, id string) {
+	err := col.FindOne(t.Context(), bson.M{"_id": id}).Decode(nil)
+	assert.ErrorIs(t, err, mongo.ErrNoDocuments)
+}
+
+func toBson(t *testing.T, v interface{}) []byte {
+	b, err := bson.Marshal(v)
+	assert.NoError(t, err)
+	return b
+}
+
+func toBsonID(t *testing.T, s string) []*adiomv1.BsonValue {
+	typ, d, err := bson.MarshalValue(s)
+	assert.NoError(t, err)
+	return []*adiomv1.BsonValue{{
+		Data: d,
+		Type: uint32(typ),
+		Name: "_id",
+	}}
+}
+
+func TestMongoConnectorUpdates(t *testing.T) {
+	client, err := MongoClient(context.Background(), ConnectorSettings{ConnectionString: TestMongoConnectionString})
+	assert.NoError(t, err)
+	col := client.Database(DBString()).Collection(ColString())
+
+	if err := col.Database().Drop(t.Context()); err != nil {
+		assert.NoError(t, err)
+	}
+	defer col.Database().Drop(t.Context())
+
+	id1 := toBsonID(t, "id1")
+	id2 := toBsonID(t, "id2")
+	id3 := toBsonID(t, "id3")
+	id4 := toBsonID(t, "id4")
+
+	conn, err := NewConn(ConnectorSettings{ConnectionString: TestMongoConnectionString, MaxPageSize: 2})
+	if err != nil {
+		assert.NoError(t, err)
+	}
+	ns := fmt.Sprintf("%s.%s", DBString(), ColString())
+
+	updateSet := []*adiomv1.Update{
+		{
+			Id:   id1,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+			Data: toBson(t, bson.M{"a": "a"}),
+		},
+		{
+			Id:   id2,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE,
+			Data: toBson(t, bson.M{"a": "b"}),
+		},
+		{
+			Id:   id3,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+			Data: toBson(t, bson.M{"a": "a"}),
+		},
+		{
+			Id:   id4,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+			Data: toBson(t, bson.M{"a": "a"}),
+		},
+		{
+			Id:   id3,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_DELETE,
+		},
+		{
+			Id:   id3,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+			Data: toBson(t, bson.M{"a": "a"}),
+		},
+		{
+			Id:   id4,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_DELETE,
+		},
+		{
+			Id:   id1,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE,
+			Data: toBson(t, bson.M{"a": "b"}),
+		},
+		{
+			Id:   id1,
+			Type: adiomv1.UpdateType_UPDATE_TYPE_INSERT,
+			Data: toBson(t, bson.M{"b": "c"}),
+		},
+	}
+
+	// Test unordered (no apply is last unless it is unique)
+	if _, err := conn.WriteUpdates(t.Context(), connect.NewRequest(&adiomv1.WriteUpdatesRequest{
+		Namespace: ns,
+		Updates:   updateSet,
+	})); err != nil {
+		assert.NoError(t, err)
+	}
+	assertDoc(t, col, map[string]string{"_id": "id1", "b": "c"})
+	assertDoc(t, col, map[string]string{"_id": "id2", "a": "b"})
+	assertDoc(t, col, map[string]string{"_id": "id3", "a": "a"})
+	assertNoDoc(t, col, "id4")
+
+	if err := col.Database().Drop(t.Context()); err != nil {
+		assert.NoError(t, err)
+	}
+
+	// Test ordered
+	if _, err := conn.WriteUpdates(t.Context(), connect.NewRequest(&adiomv1.WriteUpdatesRequest{
+		Namespace: ns,
+		Updates: append(updateSet,
+			&adiomv1.Update{
+				Id:   id1,
+				Type: adiomv1.UpdateType_UPDATE_TYPE_PARTIAL_UPDATE,
+				Data: toBson(t, bson.M{"c": "d"}),
+			},
+		),
+	})); err != nil {
+		assert.NoError(t, err)
+	}
+	assertDoc(t, col, map[string]string{"_id": "id1", "b": "c", "c": "d"})
+	assertDoc(t, col, map[string]string{"_id": "id2", "a": "b"})
+	assertDoc(t, col, map[string]string{"_id": "id3", "a": "a"})
+	assertNoDoc(t, col, "id4")
+}
+
 func TestMongoConnectorSuite2(t *testing.T) {
 	client, err := MongoClient(context.Background(), ConnectorSettings{ConnectionString: TestMongoConnectionString})
 	assert.NoError(t, err)
diff --git a/connectors/util/util.go b/connectors/util/util.go
index 48e7b2f3..c681d573 100644
--- a/connectors/util/util.go
+++ b/connectors/util/util.go
@@ -1,13 +1,36 @@
 package util
 
 import (
+	"encoding/binary"
 	"slices"
+	"strings"
 
 	adiomv1 "github.com/adiom-data/dsync/gen/adiom/v1"
 
 	"github.com/cespare/xxhash"
 )
 
+// BsonIdKey returns a stable, map-indexable key derived from a list of
+// BsonValue id parts. Name and data are length-prefixed so ids cannot
+// collide even when data contains arbitrary bytes (including nulls).
+// The returned string may contain non-UTF-8 bytes and is intended only
+// for use as a map key or for equality comparison.
+func BsonIdKey(id []*adiomv1.BsonValue) string {
+	var b strings.Builder
+	var hdr [9]byte
+	for _, p := range id {
+		name := p.GetName()
+		data := p.GetData()
+		binary.BigEndian.PutUint32(hdr[0:4], uint32(len(name)))
+		hdr[4] = byte(p.GetType())
+		binary.BigEndian.PutUint32(hdr[5:9], uint32(len(data)))
+		b.Write(hdr[:])
+		b.WriteString(name)
+		b.Write(data)
+	}
+	return b.String()
+}
+
 type dataIdIndex struct {
 	dataId []*adiomv1.BsonValue
 	index  int
diff --git a/connectors/util/util_test.go b/connectors/util/util_test.go
index cb10a1dd..7f93fe43 100644
--- a/connectors/util/util_test.go
+++ b/connectors/util/util_test.go
@@ -9,6 +9,33 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+func TestBsonIdKey(t *testing.T) {
+	// Same parts produce equal keys.
+	a := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_id", Type: 2, Data: []byte{1, 2, 3}}})
+	b := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_id", Type: 2, Data: []byte{1, 2, 3}}})
+	assert.Equal(t, a, b)
+
+	// Different data produces different keys.
+	c := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_id", Type: 2, Data: []byte{1, 2, 4}}})
+	assert.NotEqual(t, a, c)
+
+	// Different type produces different keys.
+	d := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_id", Type: 3, Data: []byte{1, 2, 3}}})
+	assert.NotEqual(t, a, d)
+
+	// Different name produces different keys.
+	e := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_ix", Type: 2, Data: []byte{1, 2, 3}}})
+	assert.NotEqual(t, a, e)
+
+	// Data containing null bytes does not collide across boundary shifts.
+	x := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_id", Type: 2, Data: []byte("ab\x00cd")}})
+	y := util.BsonIdKey([]*adiomv1.BsonValue{{Name: "_id\x00ab", Type: 2, Data: []byte("cd")}})
+	assert.NotEqual(t, x, y)
+
+	// Empty id list yields empty key.
+	assert.Equal(t, "", util.BsonIdKey(nil))
+}
+
 func TestKeepLastUpdate(t *testing.T) {
 	testData := []struct {
 		Name     string