From c4e4c20e4a35b5ef719915e85e1e706c59093d6e Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 15:09:53 +0000
Subject: [PATCH 01/13] bundle: add deployment history buffer telemetry

---
 .../bundle/dms/multiple-resources/output.txt  | 12 +++++
 .../bundle/dms/multiple-resources/script      |  4 ++
 bundle/direct/dstate/dms_test.go              | 20 ++++++++
 bundle/direct/dstate/state.go                 | 17 ++++++-
 bundle/phases/telemetry.go                    | 10 ++++
 libs/dms/operation_buffer.go                  | 44 +++++++++++++++-
 libs/dms/operation_buffer_test.go             | 51 ++++++++++++++++++-
 libs/telemetry/protos/bundle_deploy.go        | 12 +++++
 8 files changed, 167 insertions(+), 3 deletions(-)

diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index 215a2fb2534..6dc475ba046 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -10,6 +10,14 @@ Created jobs.two
 Files: 2 uploaded, 0 deleted
 Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
 
+=== History telemetry survives version completion and counts all five staged operations
+>>> cat [OUTPUT_DIR]/out.requests.txt
+{
+  "operation_count": 5,
+  "undrained_operation_count": 0,
+  "queue_size_in_range": true
+}
+
 >>> print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 {"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/1/operations/jobs.five", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[FIVE_ID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
 {"method": "PATCH", "path": "/api/2.0/bundle/deployments/[DEPLOYMENT_ID]/versions/1/operations/jobs.four", "q": {"update_mask": "state,error_message,resource_id,status"}, "body": {"error_message": "", "resource_id": "[FOUR_ID]", "sequence_id": 0, "status": "OPERATION_STATUS_SUCCEEDED"}}
@@ -172,6 +180,10 @@ Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/dms-multiple-resou
 Files: 0 uploaded, 0 deleted
 Resources: 0 created, 0 changed, 0 deleted, 5 unchanged
 
+=== A deploy with no version omits operation buffer telemetry
+>>> cat [OUTPUT_DIR]/out.requests.txt
+false
+
 >>> print_requests.py --dms //api/2.0/bundle --sort --oneline
 
 === The plan records the deployment_history feature
diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script
index 0f0276ff357..3a02a2817ff 100644
--- a/acceptance/bundle/dms/multiple-resources/script
+++ b/acceptance/bundle/dms/multiple-resources/script
@@ -2,6 +2,8 @@ envsubst < databricks.yml.tmpl > databricks.yml
 
 title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; other bundle/dms tests cover it."
 trace $CLI bundle deploy
+title "History telemetry survives version completion and counts all five staged operations"
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, undrained_operation_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
 replace_ids.py
 trace print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 
@@ -18,6 +20,8 @@ trace print_state.py --no-dms
 
 title "Redeploy with no changes: nothing is applied, so no version is created at all - the deployment's version tracks state changes, like serial"
 trace $CLI bundle deploy
+title "A deploy with no version omits operation buffer telemetry"
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event | has("deployment_history")'
 trace print_requests.py --dms //api/2.0/bundle --sort --oneline
 
 title "The plan records the deployment_history feature"
diff --git a/bundle/direct/dstate/dms_test.go b/bundle/direct/dstate/dms_test.go
index ab4e1bc017f..8ec404616d0 100644
--- a/bundle/direct/dstate/dms_test.go
+++ b/bundle/direct/dstate/dms_test.go
@@ -10,6 +10,26 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestOperationBufferStatsSurviveVersionCompletion(t *testing.T) {
+	var db DeploymentState
+	assert.Nil(t, db.OperationBufferStats())
+	db.InitializeOperationBuffer(t.Context(), "dep-1", 1, 5)
+	require.NoError(t, db.operationBuffer.Drain(t.Context()))
+	want := &dms.OperationBufferStats{OperationCount: 5}
+	assert.Equal(t, want, db.OperationBufferStats())
+
+	db.mu.Lock()
+	db.reset()
+	db.versionCompleted = true
+	db.reset()
+	db.mu.Unlock()
+	assert.Nil(t, db.operationBuffer)
+	assert.Equal(t, want, db.OperationBufferStats())
+
+	require.NoError(t, db.Open(t.Context(), t.TempDir()+"/resources.json", ReadingFile, WithRecovery(false), OpenDmsArgs{}))
+	assert.Nil(t, db.OperationBufferStats())
+}
+
 func TestApplyDMSState(t *testing.T) {
 	// The service stores state as an opaque string, so the envelope the write path sent
 	// arrives verbatim.
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index 384246913f4..138e9eb3519 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -105,7 +105,8 @@ type DeploymentState struct {
 
 	// operationBuffer records each state write with DMS. Nil unless the bundle records deployment
 	// history, in which case InitializeOperationBuffer installs it once the version exists.
-	operationBuffer *dms.OperationBuffer
+	operationBuffer      *dms.OperationBuffer
+	operationBufferStats *dms.OperationBufferStats
 
 	// versionCompleted makes CompleteVersion a no-op after the first call, so a deferred safety-net
 	// completion after an explicit one does nothing.
@@ -199,6 +200,17 @@ func (db *DeploymentState) getOperationBuffer() *dms.OperationBuffer {
 	return db.operationBuffer
 }
 
+// OperationBufferStats returns history measurements, including after version completion.
+func (db *DeploymentState) OperationBufferStats() *dms.OperationBufferStats {
+	db.mu.Lock()
+	defer db.mu.Unlock()
+	if db.operationBuffer != nil {
+		stats := db.operationBuffer.Stats()
+		return &stats
+	}
+	return db.operationBufferStats
+}
+
 // RecordingError reports whether recording state writes to the service has failed, so the apply
 // stops touching resources once the service is no longer keeping up. File-backed history modes
 // always return nil so history recording errors do not block the deployment; resource state
@@ -562,6 +574,8 @@ func (db *DeploymentState) reset() {
 	// Telemetry and destroy still need the mode and IDs after version completion.
 	// Clear the recording buffer once CompleteVersion has copied it.
 	if db.versionCompleted {
+		stats := db.operationBuffer.Stats()
+		db.operationBufferStats = &stats
 		db.operationBuffer = nil
 		db.versionCompleted = false
 	}
@@ -569,6 +583,7 @@ func (db *DeploymentState) reset() {
 
 func (db *DeploymentState) unlockedOpen(ctx context.Context, path string, mode Mode, withRecovery WithRecovery, dmsDeployment OpenDmsArgs) error {
 	db.Path = path
+	db.operationBufferStats = nil
 	db.mode = mode
 	db.DeploymentID = dmsDeployment.DeploymentID
 	db.VersionID = dmsDeployment.LastVersionID
diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go
index 2e6296be7f8..cd35874a2e8 100644
--- a/bundle/phases/telemetry.go
+++ b/bundle/phases/telemetry.go
@@ -404,6 +404,15 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 		uploadFileSizers[i] = f
 	}
 
+	var deploymentHistory *protos.BundleDeploymentHistory
+	if stats := b.DeploymentBundle.StateDB.OperationBufferStats(); stats != nil {
+		deploymentHistory = &protos.BundleDeploymentHistory{
+			OperationCount:          int64(stats.OperationCount),
+			MaxQueueSize:            int64(stats.MaxQueueSize),
+			UndrainedOperationCount: int64(stats.UndrainedOperationCount),
+		}
+	}
+
 	telemetry.Log(ctx, protos.DatabricksCliLog{
 		BundleDeployEvent: &protos.BundleDeployEvent{
 			BundleUuid:   bundleUuid,
@@ -430,6 +439,7 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 			ResourceDashboardIDs: dashboardIds,
 
 			ResourcesMetadata: collectResourcesMetadata(ctx, b),
+			DeploymentHistory: deploymentHistory,
 
 			Pydabs: pydabsSection,
 
diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 7a77cdf83c8..7bd0b86d264 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -25,6 +25,13 @@ type BufferOptions struct {
 	OperationCount int
 }
 
+// OperationBufferStats describes one deployment version's history uploads.
+type OperationBufferStats struct {
+	OperationCount          int
+	MaxQueueSize            int
+	UndrainedOperationCount int
+}
+
 // OperationBuffer records each state write with DMS for one deployment version, off the apply
 // path: writes are queued and sent on one background goroutine. It exists only while a bundle
 // records deployment history; callers hold a nil buffer otherwise and must not call it.
@@ -56,7 +63,10 @@ type OperationBuffer struct {
 	mu sync.Mutex
 
 	// pending holds the newest update per resource key, absent once the writer takes it.
-	pending map[string]OperationUpdate
+	pending            map[string]OperationUpdate
+	inFlight           string
+	stats              OperationBufferStats
+	drainStatsRecorded bool
 
 	// latestState is the newest state-bearing update per resource key, kept for the buffer's
 	// whole life - unlike pending, which the writer removes on take. RecordFailure merges its
@@ -90,6 +100,7 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		pending:      make(map[string]OperationUpdate),
 		latestState:  make(map[string]OperationUpdate),
 		sequenceIDs:  make(map[string]int64),
+		stats:        OperationBufferStats{OperationCount: options.OperationCount},
 	}
 	b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
 	go b.run(ctx)
@@ -133,6 +144,7 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 		update = waiting.Merge(update)
 	}
 	b.pending[resourceKey] = update
+	b.stats.MaxQueueSize = max(b.stats.MaxQueueSize, len(b.pending))
 	b.mu.Unlock()
 
 	// when already queued: the writer reads the map when it gets to the key,
@@ -151,6 +163,9 @@ func (b *OperationBuffer) take(resourceKey string) (OperationUpdate, bool) {
 
 	update, ok := b.pending[resourceKey]
 	delete(b.pending, resourceKey)
+	if ok {
+		b.inFlight = resourceKey
+	}
 	return update, ok
 }
 
@@ -186,6 +201,9 @@ func (b *OperationBuffer) run(ctx context.Context) {
 		if err := b.write(ctx, resourceKey, update); err != nil {
 			b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		}
+		b.mu.Lock()
+		b.inFlight = ""
+		b.mu.Unlock()
 	}
 }
 
@@ -217,6 +235,7 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 	if !b.bestEffort {
 		// DMS is authoritative: wait for all queued writes and return any recording error.
 		<-b.done
+		b.recordDrainStats()
 		return b.Err()
 	}
 
@@ -227,15 +246,38 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 	select {
 	case <-b.done:
 		// Report the first upload error even if later uploads succeeded.
+		b.recordDrainStats()
 		return b.Err()
 	case <-ctx.Done():
 		// Deadline or cancellation: stop the upload so deployment can finish promptly.
+		b.recordDrainStats()
 		b.setErr(ctx.Err())
 		b.cancel()
 		return b.Err()
 	}
 }
 
+// recordDrainStats freezes outstanding resource counts before cancellation can clear an upload.
+func (b *OperationBuffer) recordDrainStats() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.drainStatsRecorded {
+		return
+	}
+	b.drainStatsRecorded = true
+	b.stats.UndrainedOperationCount = len(b.pending)
+	if _, queued := b.pending[b.inFlight]; b.inFlight != "" && !queued {
+		b.stats.UndrainedOperationCount++
+	}
+}
+
+// Stats returns a snapshot; undrained operations are counted once when Drain ends.
+func (b *OperationBuffer) Stats() OperationBufferStats {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.stats
+}
+
 // setErr keeps the first recording error for Drain to report.
 func (b *OperationBuffer) setErr(err error) {
 	b.mu.Lock()
diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go
index 0e019284360..60fb235a123 100644
--- a/libs/dms/operation_buffer_test.go
+++ b/libs/dms/operation_buffer_test.go
@@ -74,9 +74,58 @@ func TestBufferFailsOnOversizedState(t *testing.T) {
 
 func TestDrainIsIdempotent(t *testing.T) {
 	// An idle writer must finish when draining begins.
-	b := StartOperationBuffer(t.Context(), "dep-1", 1, BufferOptions{})
+	b := StartOperationBuffer(t.Context(), "dep-1", 1, BufferOptions{OperationCount: 5})
 	require.NoError(t, b.Drain(t.Context()))
 	require.NoError(t, b.Drain(t.Context()))
+	assert.Equal(t, OperationBufferStats{OperationCount: 5}, b.Stats())
+}
+
+func TestBufferMaxQueueSize(t *testing.T) {
+	b := queued()
+	for _, key := range []string{"a", "b", "a", "c"} {
+		b.record(key, stateUpdate(t, key), false, nil)
+	}
+	assert.Equal(t, 3, b.Stats().MaxQueueSize)
+	_, ok := b.take("a")
+	require.True(t, ok)
+	b.RecordFailure("b", "id-1", nil, errors.New("apply failed"))
+	assert.Equal(t, 3, b.Stats().MaxQueueSize)
+}
+
+func TestHistoryDrainStats(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		inFlight string
+		want     int
+	}{
+		{name: "queued", want: 2},
+		{name: "in flight", inFlight: "c", want: 3},
+		{name: "in flight and queued for the same operation", inFlight: "a", want: 2},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			synctest.Test(t, func(t *testing.T) {
+				b := queued()
+				b.bestEffort = true
+				_, b.cancel = context.WithCancel(t.Context())
+				b.drainTimeout = time.Second
+				b.done = make(chan struct{})
+				b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
+				b.record("a", stateUpdate(t, "a"), false, nil)
+				b.record("b", stateUpdate(t, "b"), false, nil)
+				b.inFlight = tc.inFlight
+
+				require.ErrorIs(t, b.Drain(t.Context()), context.DeadlineExceeded)
+				assert.Equal(t, tc.want, b.Stats().UndrainedOperationCount)
+
+				// Cancellation may finish the upload after Drain returns.
+				b.inFlight = ""
+				clear(b.pending)
+				close(b.done)
+				require.ErrorIs(t, b.Drain(t.Context()), context.DeadlineExceeded)
+				assert.Equal(t, tc.want, b.Stats().UndrainedOperationCount)
+			})
+		})
+	}
 }
 
 func TestRecordFailureKeepsTheStateTheLastWriteRecorded(t *testing.T) {
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index 4464f36bca4..67ef9042b18 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -38,6 +38,8 @@ type BundleDeployEvent struct {
 	// Per-resource-type metadata (counts and state-size statistics).
 	ResourcesMetadata *BundleResourcesMetadata `json:"resources_metadata,omitempty"`
 
+	DeploymentHistory *BundleDeploymentHistory `json:"deployment_history,omitempty"`
+
 	// Usage of Python support (PyDABs). Set only when the bundle declares
 	// Python resource loaders or mutators.
 	Pydabs *BundleDeployPydabs `json:"pydabs,omitempty"`
@@ -45,6 +47,16 @@ type BundleDeployEvent struct {
 	Experimental *BundleDeployExperimental `json:"experimental,omitempty"`
 }
 
+// BundleDeploymentHistory measures history recording for one deployment version.
+type BundleDeploymentHistory struct {
+	// Number of resource operations staged for this deployment version.
+	OperationCount int64 `json:"operation_count"`
+	// Peak number of queued resource keys, excluding the in-flight upload.
+	MaxQueueSize int64 `json:"max_queue_size"`
+	// Distinct resource operations still queued or in flight when Drain ends.
+	UndrainedOperationCount int64 `json:"undrained_operation_count"`
+}
+
 // These metrics are experimental and are often added in an adhoc manner. There
 // are no guarantees for these metrics and they maybe removed in the future without
 // any notice.

From a78eec5a3c92c5c30f5cfad654a47fc939508ca5 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 16:08:02 +0000
Subject: [PATCH 02/13] bundle: inline deployment history drain measurements

---
 libs/dms/operation_buffer.go | 34 ++++++++++++++--------------------
 1 file changed, 14 insertions(+), 20 deletions(-)

diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 7bd0b86d264..1a5c35a8dfb 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -63,10 +63,9 @@ type OperationBuffer struct {
 	mu sync.Mutex
 
 	// pending holds the newest update per resource key, absent once the writer takes it.
-	pending            map[string]OperationUpdate
-	inFlight           string
-	stats              OperationBufferStats
-	drainStatsRecorded bool
+	pending  map[string]OperationUpdate
+	inFlight string
+	stats    OperationBufferStats
 
 	// latestState is the newest state-bearing update per resource key, kept for the buffer's
 	// whole life - unlike pending, which the writer removes on take. RecordFailure merges its
@@ -235,7 +234,6 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 	if !b.bestEffort {
 		// DMS is authoritative: wait for all queued writes and return any recording error.
 		<-b.done
-		b.recordDrainStats()
 		return b.Err()
 	}
 
@@ -243,32 +241,28 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 	// including operations already uploaded; caller cancellation can shorten the wait.
 	ctx, cancel := context.WithTimeout(ctx, b.drainTimeout)
 	defer cancel()
+	var drainErr error
 	select {
 	case <-b.done:
 		// Report the first upload error even if later uploads succeeded.
-		b.recordDrainStats()
-		return b.Err()
 	case <-ctx.Done():
 		// Deadline or cancellation: stop the upload so deployment can finish promptly.
-		b.recordDrainStats()
-		b.setErr(ctx.Err())
-		b.cancel()
-		return b.Err()
+		drainErr = ctx.Err()
 	}
-}
 
-// recordDrainStats freezes outstanding resource counts before cancellation can clear an upload.
-func (b *OperationBuffer) recordDrainStats() {
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	if b.drainStatsRecorded {
-		return
-	}
-	b.drainStatsRecorded = true
-	b.stats.UndrainedOperationCount = len(b.pending)
+	undrained := len(b.pending)
 	if _, queued := b.pending[b.inFlight]; b.inFlight != "" && !queued {
-		b.stats.UndrainedOperationCount++
+		undrained++
+	}
+	// No writes can be queued after Drain starts, so retain the first drain's count.
+	b.stats.UndrainedOperationCount = max(b.stats.UndrainedOperationCount, undrained)
+	if b.err == nil {
+		b.err = drainErr
 	}
+	b.cancel()
+	return b.err
 }
 
 // Stats returns a snapshot; undrained operations are counted once when Drain ends.

From 87ba39a15d52961bd2439cdb3f0649aba0aa92a9 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 16:58:24 +0000
Subject: [PATCH 03/13] bundle: simplify deployment history backlog telemetry

---
 libs/dms/operation_buffer.go           | 17 +++--------------
 libs/dms/operation_buffer_test.go      | 25 +++++++++++++++++--------
 libs/telemetry/protos/bundle_deploy.go |  2 +-
 3 files changed, 21 insertions(+), 23 deletions(-)

diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 1a5c35a8dfb..2c8ac1bd151 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -63,9 +63,8 @@ type OperationBuffer struct {
 	mu sync.Mutex
 
 	// pending holds the newest update per resource key, absent once the writer takes it.
-	pending  map[string]OperationUpdate
-	inFlight string
-	stats    OperationBufferStats
+	pending map[string]OperationUpdate
+	stats   OperationBufferStats
 
 	// latestState is the newest state-bearing update per resource key, kept for the buffer's
 	// whole life - unlike pending, which the writer removes on take. RecordFailure merges its
@@ -162,9 +161,6 @@ func (b *OperationBuffer) take(resourceKey string) (OperationUpdate, bool) {
 
 	update, ok := b.pending[resourceKey]
 	delete(b.pending, resourceKey)
-	if ok {
-		b.inFlight = resourceKey
-	}
 	return update, ok
 }
 
@@ -200,9 +196,6 @@ func (b *OperationBuffer) run(ctx context.Context) {
 		if err := b.write(ctx, resourceKey, update); err != nil {
 			b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		}
-		b.mu.Lock()
-		b.inFlight = ""
-		b.mu.Unlock()
 	}
 }
 
@@ -252,12 +245,8 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	undrained := len(b.pending)
-	if _, queued := b.pending[b.inFlight]; b.inFlight != "" && !queued {
-		undrained++
-	}
 	// No writes can be queued after Drain starts, so retain the first drain's count.
-	b.stats.UndrainedOperationCount = max(b.stats.UndrainedOperationCount, undrained)
+	b.stats.UndrainedOperationCount = max(b.stats.UndrainedOperationCount, len(b.queue))
 	if b.err == nil {
 		b.err = drainErr
 	}
diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go
index 60fb235a123..8d106e08090 100644
--- a/libs/dms/operation_buffer_test.go
+++ b/libs/dms/operation_buffer_test.go
@@ -94,13 +94,14 @@ func TestBufferMaxQueueSize(t *testing.T) {
 
 func TestHistoryDrainStats(t *testing.T) {
 	for _, tc := range []struct {
-		name     string
-		inFlight string
-		want     int
+		name    string
+		take    bool
+		requeue bool
+		want    int
 	}{
 		{name: "queued", want: 2},
-		{name: "in flight", inFlight: "c", want: 3},
-		{name: "in flight and queued for the same operation", inFlight: "a", want: 2},
+		{name: "active upload is excluded", take: true, want: 1},
+		{name: "new update queued for the active resource", take: true, requeue: true, want: 2},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			synctest.Test(t, func(t *testing.T) {
@@ -112,13 +113,21 @@ func TestHistoryDrainStats(t *testing.T) {
 				b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
 				b.record("a", stateUpdate(t, "a"), false, nil)
 				b.record("b", stateUpdate(t, "b"), false, nil)
-				b.inFlight = tc.inFlight
+				if tc.take {
+					key := <-b.queue
+					_, ok := b.take(key)
+					require.True(t, ok)
+					if tc.requeue {
+						b.record(key, stateUpdate(t, key), false, nil)
+					}
+				}
 
 				require.ErrorIs(t, b.Drain(t.Context()), context.DeadlineExceeded)
 				assert.Equal(t, tc.want, b.Stats().UndrainedOperationCount)
 
-				// Cancellation may finish the upload after Drain returns.
-				b.inFlight = ""
+				// The worker may empty the queue after Drain returns.
+				for range b.queue {
+				}
 				clear(b.pending)
 				close(b.done)
 				require.ErrorIs(t, b.Drain(t.Context()), context.DeadlineExceeded)
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index 67ef9042b18..bc781b0a708 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -53,7 +53,7 @@ type BundleDeploymentHistory struct {
 	OperationCount int64 `json:"operation_count"`
 	// Peak number of queued resource keys, excluding the in-flight upload.
 	MaxQueueSize int64 `json:"max_queue_size"`
-	// Distinct resource operations still queued or in flight when Drain ends.
+	// Resource operations still queued when Drain ends, excluding the active upload.
 	UndrainedOperationCount int64 `json:"undrained_operation_count"`
 }
 

From 6032d50497390aac7015c34d13f4d6cc9b83169f Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 17:13:16 +0000
Subject: [PATCH 04/13] bundle: record undrained operations only when drain is
 interrupted

---
 libs/dms/operation_buffer.go | 22 ++++++++++------------
 1 file changed, 10 insertions(+), 12 deletions(-)

diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 2c8ac1bd151..6c7fad4d69c 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -234,24 +234,22 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 	// including operations already uploaded; caller cancellation can shorten the wait.
 	ctx, cancel := context.WithTimeout(ctx, b.drainTimeout)
 	defer cancel()
-	var drainErr error
 	select {
 	case <-b.done:
 		// Report the first upload error even if later uploads succeeded.
+		return b.Err()
 	case <-ctx.Done():
 		// Deadline or cancellation: stop the upload so deployment can finish promptly.
-		drainErr = ctx.Err()
-	}
-
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	// No writes can be queued after Drain starts, so retain the first drain's count.
-	b.stats.UndrainedOperationCount = max(b.stats.UndrainedOperationCount, len(b.queue))
-	if b.err == nil {
-		b.err = drainErr
+		b.mu.Lock()
+		defer b.mu.Unlock()
+		// No writes can be queued after Drain starts, so retain the first drain's count.
+		b.stats.UndrainedOperationCount = max(b.stats.UndrainedOperationCount, len(b.queue))
+		if b.err == nil {
+			b.err = ctx.Err()
+		}
+		b.cancel()
+		return b.err
 	}
-	b.cancel()
-	return b.err
 }
 
 // Stats returns a snapshot; undrained operations are counted once when Drain ends.

From 935708584dde1f205f46bf2d006ead6bea63dc39 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 17:17:31 +0000
Subject: [PATCH 05/13] bundle: directly record queue length on drain timeout

---
 libs/dms/operation_buffer.go | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 6c7fad4d69c..92a0c5edda7 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -242,8 +242,7 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 		// Deadline or cancellation: stop the upload so deployment can finish promptly.
 		b.mu.Lock()
 		defer b.mu.Unlock()
-		// No writes can be queued after Drain starts, so retain the first drain's count.
-		b.stats.UndrainedOperationCount = max(b.stats.UndrainedOperationCount, len(b.queue))
+		b.stats.UndrainedOperationCount = len(b.queue)
 		if b.err == nil {
 			b.err = ctx.Err()
 		}
@@ -252,7 +251,7 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 	}
 }
 
-// Stats returns a snapshot; undrained operations are counted once when Drain ends.
+// Stats returns a snapshot of the history measurements.
 func (b *OperationBuffer) Stats() OperationBufferStats {
 	b.mu.Lock()
 	defer b.mu.Unlock()

From 2b4a818db63037e0f4c10016462c1f21fcf9e24e Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 22:22:50 +0000
Subject: [PATCH 06/13] bundle: drop operation buffer telemetry unit tests

---
 libs/dms/operation_buffer_test.go | 60 +------------------------------
 1 file changed, 1 insertion(+), 59 deletions(-)

diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go
index 8d106e08090..0e019284360 100644
--- a/libs/dms/operation_buffer_test.go
+++ b/libs/dms/operation_buffer_test.go
@@ -74,67 +74,9 @@ func TestBufferFailsOnOversizedState(t *testing.T) {
 
 func TestDrainIsIdempotent(t *testing.T) {
 	// An idle writer must finish when draining begins.
-	b := StartOperationBuffer(t.Context(), "dep-1", 1, BufferOptions{OperationCount: 5})
+	b := StartOperationBuffer(t.Context(), "dep-1", 1, BufferOptions{})
 	require.NoError(t, b.Drain(t.Context()))
 	require.NoError(t, b.Drain(t.Context()))
-	assert.Equal(t, OperationBufferStats{OperationCount: 5}, b.Stats())
-}
-
-func TestBufferMaxQueueSize(t *testing.T) {
-	b := queued()
-	for _, key := range []string{"a", "b", "a", "c"} {
-		b.record(key, stateUpdate(t, key), false, nil)
-	}
-	assert.Equal(t, 3, b.Stats().MaxQueueSize)
-	_, ok := b.take("a")
-	require.True(t, ok)
-	b.RecordFailure("b", "id-1", nil, errors.New("apply failed"))
-	assert.Equal(t, 3, b.Stats().MaxQueueSize)
-}
-
-func TestHistoryDrainStats(t *testing.T) {
-	for _, tc := range []struct {
-		name    string
-		take    bool
-		requeue bool
-		want    int
-	}{
-		{name: "queued", want: 2},
-		{name: "active upload is excluded", take: true, want: 1},
-		{name: "new update queued for the active resource", take: true, requeue: true, want: 2},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			synctest.Test(t, func(t *testing.T) {
-				b := queued()
-				b.bestEffort = true
-				_, b.cancel = context.WithCancel(t.Context())
-				b.drainTimeout = time.Second
-				b.done = make(chan struct{})
-				b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
-				b.record("a", stateUpdate(t, "a"), false, nil)
-				b.record("b", stateUpdate(t, "b"), false, nil)
-				if tc.take {
-					key := <-b.queue
-					_, ok := b.take(key)
-					require.True(t, ok)
-					if tc.requeue {
-						b.record(key, stateUpdate(t, key), false, nil)
-					}
-				}
-
-				require.ErrorIs(t, b.Drain(t.Context()), context.DeadlineExceeded)
-				assert.Equal(t, tc.want, b.Stats().UndrainedOperationCount)
-
-				// The worker may empty the queue after Drain returns.
-				for range b.queue {
-				}
-				clear(b.pending)
-				close(b.done)
-				require.ErrorIs(t, b.Drain(t.Context()), context.DeadlineExceeded)
-				assert.Equal(t, tc.want, b.Stats().UndrainedOperationCount)
-			})
-		})
-	}
 }
 
 func TestRecordFailureKeepsTheStateTheLastWriteRecorded(t *testing.T) {

From e36fb9ebd86a743fc3cb75ce3a9541fd6842b51b Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 22:23:57 +0000
Subject: [PATCH 07/13] bundle: drop deployment state telemetry unit test

---
 bundle/direct/dstate/dms_test.go | 20 --------------------
 1 file changed, 20 deletions(-)

diff --git a/bundle/direct/dstate/dms_test.go b/bundle/direct/dstate/dms_test.go
index 8ec404616d0..ab4e1bc017f 100644
--- a/bundle/direct/dstate/dms_test.go
+++ b/bundle/direct/dstate/dms_test.go
@@ -10,26 +10,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestOperationBufferStatsSurviveVersionCompletion(t *testing.T) {
-	var db DeploymentState
-	assert.Nil(t, db.OperationBufferStats())
-	db.InitializeOperationBuffer(t.Context(), "dep-1", 1, 5)
-	require.NoError(t, db.operationBuffer.Drain(t.Context()))
-	want := &dms.OperationBufferStats{OperationCount: 5}
-	assert.Equal(t, want, db.OperationBufferStats())
-
-	db.mu.Lock()
-	db.reset()
-	db.versionCompleted = true
-	db.reset()
-	db.mu.Unlock()
-	assert.Nil(t, db.operationBuffer)
-	assert.Equal(t, want, db.OperationBufferStats())
-
-	require.NoError(t, db.Open(t.Context(), t.TempDir()+"/resources.json", ReadingFile, WithRecovery(false), OpenDmsArgs{}))
-	assert.Nil(t, db.OperationBufferStats())
-}
-
 func TestApplyDMSState(t *testing.T) {
 	// The service stores state as an opaque string, so the envelope the write path sent
 	// arrives verbatim.

From 62611f87edaa14d6ce6968f2a779b2688640c3a7 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 22:33:15 +0000
Subject: [PATCH 08/13] bundle: track deployment history recording errors

---
 .../bundle/dms/multiple-resources/output.txt       |  1 +
 acceptance/bundle/dms/multiple-resources/script    |  2 +-
 .../dms/operation-upload-fails-once/output.txt     |  7 +++++++
 .../bundle/dms/operation-upload-fails-once/script  |  2 ++
 bundle/phases/telemetry.go                         |  1 +
 libs/dms/operation_buffer.go                       | 14 +++++++++-----
 libs/telemetry/protos/bundle_deploy.go             |  3 +++
 7 files changed, 24 insertions(+), 6 deletions(-)

diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index 6dc475ba046..18ef4e3ccec 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -15,6 +15,7 @@ Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
 {
   "operation_count": 5,
   "undrained_operation_count": 0,
+  "recording_error_count": 0,
   "queue_size_in_range": true
 }
 
diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script
index 3a02a2817ff..997e3de4ca3 100644
--- a/acceptance/bundle/dms/multiple-resources/script
+++ b/acceptance/bundle/dms/multiple-resources/script
@@ -3,7 +3,7 @@ envsubst < databricks.yml.tmpl > databricks.yml
 title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; other bundle/dms tests cover it."
 trace $CLI bundle deploy
 title "History telemetry survives version completion and counts all five staged operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, undrained_operation_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, undrained_operation_count, recording_error_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
 replace_ids.py
 trace print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index d17e9ee26fb..26db57c06d2 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -68,6 +68,13 @@ Warn: Failed to record deployment history: recording operation for resources.sch
 Files: 2 uploaded, 0 deleted
 Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
 
+=== A failed history upload is counted separately from the queued operations
+>>> cat [OUTPUT_DIR]/out.requests.txt
+{
+  "recording_error_count": 1,
+  "undrained_operation_count": 0
+}
+
 === The service lists the schema with the failure
 >>> MSYS_NO_PATHCONV=1 [CLI] bundle-deployments list-resources deployments/[DEPLOYMENT_ID] -o json
 [
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script
index 69dab957164..3cb6795da5a 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/script
+++ b/acceptance/bundle/dms/operation-upload-fails-once/script
@@ -23,6 +23,8 @@ $CLI catalogs create "$other_catalog" &> LOG.catalog
 trap '$CLI catalogs delete "$other_catalog" --force >> LOG.catalog 2>&1' EXIT
 trace update_file.py databricks.yml "catalog_name: main" "catalog_name: ${other_catalog}"
 trace $CLI bundle deploy --auto-approve
+title "A failed history upload is counted separately from the queued operations"
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {recording_error_count, undrained_operation_count}'
 
 title "The service lists the schema with the failure"
 deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-upload-fails-once-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go
index cd35874a2e8..13e14757e64 100644
--- a/bundle/phases/telemetry.go
+++ b/bundle/phases/telemetry.go
@@ -410,6 +410,7 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 			OperationCount:          int64(stats.OperationCount),
 			MaxQueueSize:            int64(stats.MaxQueueSize),
 			UndrainedOperationCount: int64(stats.UndrainedOperationCount),
+			RecordingErrorCount:     int64(stats.RecordingErrorCount),
 		}
 	}
 
diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 92a0c5edda7..57dc16e88a5 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -30,6 +30,7 @@ type OperationBufferStats struct {
 	OperationCount          int
 	MaxQueueSize            int
 	UndrainedOperationCount int
+	RecordingErrorCount     int
 }
 
 // OperationBuffer records each state write with DMS for one deployment version, off the apply
@@ -110,7 +111,7 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 func (b *OperationBuffer) RecordOperation(ctx context.Context, resourceKey string, inProgress bool, resourceID string, state json.RawMessage) {
 	update, err := NewStateUpdate(resourceID, state, inProgress)
 	if err != nil {
-		b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
+		b.setErr(ctx, fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		return
 	}
 
@@ -176,7 +177,7 @@ func (b *OperationBuffer) run(ctx context.Context) {
 			case resourceKey, ok = <-b.queue:
 			case <-ctx.Done():
 				// Drain calls cancel on timeout. Parent cancellation can also stop history.
-				b.setErr(ctx.Err())
+				b.setErr(ctx, ctx.Err())
 				return
 			}
 		} else {
@@ -194,7 +195,7 @@ func (b *OperationBuffer) run(ctx context.Context) {
 
 		// Keep going after a failure, so one bad write does not drop everything behind it.
 		if err := b.write(ctx, resourceKey, update); err != nil {
-			b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
+			b.setErr(ctx, fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		}
 	}
 }
@@ -258,11 +259,14 @@ func (b *OperationBuffer) Stats() OperationBufferStats {
 	return b.stats
 }
 
-// setErr keeps the first recording error for Drain to report.
-func (b *OperationBuffer) setErr(err error) {
+// setErr counts recording errors while ctx is active and keeps the first error for Drain.
+func (b *OperationBuffer) setErr(ctx context.Context, err error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if ctx.Err() == nil {
+		b.stats.RecordingErrorCount++
+	}
 	if b.err == nil {
 		b.err = err
 	}
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index bc781b0a708..7f3707b7fa1 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -55,6 +55,9 @@ type BundleDeploymentHistory struct {
 	MaxQueueSize int64 `json:"max_queue_size"`
 	// Resource operations still queued when Drain ends, excluding the active upload.
 	UndrainedOperationCount int64 `json:"undrained_operation_count"`
+	// Failed update-building or upload attempts, excluding context cancellation.
+	// Later successful updates do not clear the count.
+	RecordingErrorCount int64 `json:"recording_error_count"`
 }
 
 // These metrics are experimental and are often added in an adhoc manner. There

From 8ee260720e5754d041f430aa74ef3444e5931c42 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 22:55:37 +0000
Subject: [PATCH 09/13] bundle: count history upload successes and errors at
 the API call

---
 .../bundle/dms/multiple-resources/output.txt  |  1 +
 .../bundle/dms/multiple-resources/script      |  2 +-
 .../operation-upload-fails-once/output.txt    |  1 +
 .../dms/operation-upload-fails-once/script    |  2 +-
 bundle/phases/telemetry.go                    |  1 +
 libs/dms/operation_buffer.go                  | 21 ++++++++++++-------
 libs/telemetry/protos/bundle_deploy.go        |  4 +++-
 7 files changed, 21 insertions(+), 11 deletions(-)

diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index 18ef4e3ccec..7d139818038 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -16,6 +16,7 @@ Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
   "operation_count": 5,
   "undrained_operation_count": 0,
   "recording_error_count": 0,
+  "recording_success_count": 5,
   "queue_size_in_range": true
 }
 
diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script
index 997e3de4ca3..76c027bd559 100644
--- a/acceptance/bundle/dms/multiple-resources/script
+++ b/acceptance/bundle/dms/multiple-resources/script
@@ -3,7 +3,7 @@ envsubst < databricks.yml.tmpl > databricks.yml
 title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; other bundle/dms tests cover it."
 trace $CLI bundle deploy
 title "History telemetry survives version completion and counts all five staged operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, undrained_operation_count, recording_error_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, undrained_operation_count, recording_error_count, recording_success_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
 replace_ids.py
 trace print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index 26db57c06d2..da8d71fec48 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -72,6 +72,7 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
 >>> cat [OUTPUT_DIR]/out.requests.txt
 {
   "recording_error_count": 1,
+  "recording_success_count": 1,
   "undrained_operation_count": 0
 }
 
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script
index 3cb6795da5a..977c7b04898 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/script
+++ b/acceptance/bundle/dms/operation-upload-fails-once/script
@@ -24,7 +24,7 @@ trap '$CLI catalogs delete "$other_catalog" --force >> LOG.catalog 2>&1' EXIT
 trace update_file.py databricks.yml "catalog_name: main" "catalog_name: ${other_catalog}"
 trace $CLI bundle deploy --auto-approve
 title "A failed history upload is counted separately from the queued operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {recording_error_count, undrained_operation_count}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {recording_error_count, recording_success_count, undrained_operation_count}'
 
 title "The service lists the schema with the failure"
 deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-upload-fails-once-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go
index 13e14757e64..1b1fa41e54a 100644
--- a/bundle/phases/telemetry.go
+++ b/bundle/phases/telemetry.go
@@ -411,6 +411,7 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 			MaxQueueSize:            int64(stats.MaxQueueSize),
 			UndrainedOperationCount: int64(stats.UndrainedOperationCount),
 			RecordingErrorCount:     int64(stats.RecordingErrorCount),
+			RecordingSuccessCount:   int64(stats.RecordingSuccessCount),
 		}
 	}
 
diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 57dc16e88a5..9f93a6695d9 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -31,6 +31,7 @@ type OperationBufferStats struct {
 	MaxQueueSize            int
 	UndrainedOperationCount int
 	RecordingErrorCount     int
+	RecordingSuccessCount   int
 }
 
 // OperationBuffer records each state write with DMS for one deployment version, off the apply
@@ -111,7 +112,10 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 func (b *OperationBuffer) RecordOperation(ctx context.Context, resourceKey string, inProgress bool, resourceID string, state json.RawMessage) {
 	update, err := NewStateUpdate(resourceID, state, inProgress)
 	if err != nil {
-		b.setErr(ctx, fmt.Errorf("recording operation for %s: %w", resourceKey, err))
+		b.mu.Lock()
+		b.stats.RecordingErrorCount++
+		b.mu.Unlock()
+		b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		return
 	}
 
@@ -177,7 +181,7 @@ func (b *OperationBuffer) run(ctx context.Context) {
 			case resourceKey, ok = <-b.queue:
 			case <-ctx.Done():
 				// Drain calls cancel on timeout. Parent cancellation can also stop history.
-				b.setErr(ctx, ctx.Err())
+				b.setErr(ctx.Err())
 				return
 			}
 		} else {
@@ -195,7 +199,7 @@ func (b *OperationBuffer) run(ctx context.Context) {
 
 		// Keep going after a failure, so one bad write does not drop everything behind it.
 		if err := b.write(ctx, resourceKey, update); err != nil {
-			b.setErr(ctx, fmt.Errorf("recording operation for %s: %w", resourceKey, err))
+			b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		}
 	}
 }
@@ -208,9 +212,13 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio
 		Operation:  update.operation(b.sequenceIDs[key]),
 		UpdateMask: update.Fields.Mask(),
 	})
+	b.mu.Lock()
+	defer b.mu.Unlock()
 	if err != nil {
+		b.stats.RecordingErrorCount++
 		return err
 	}
+	b.stats.RecordingSuccessCount++
 
 	// The next write for this resource echoes the sequence id this one earned.
 	b.sequenceIDs[key] = result.SequenceId
@@ -259,14 +267,11 @@ func (b *OperationBuffer) Stats() OperationBufferStats {
 	return b.stats
 }
 
-// setErr counts recording errors while ctx is active and keeps the first error for Drain.
-func (b *OperationBuffer) setErr(ctx context.Context, err error) {
+// setErr keeps the first recording error for Drain to report.
+func (b *OperationBuffer) setErr(err error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	if ctx.Err() == nil {
-		b.stats.RecordingErrorCount++
-	}
 	if b.err == nil {
 		b.err = err
 	}
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index 7f3707b7fa1..bd8659ca37c 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -55,9 +55,11 @@ type BundleDeploymentHistory struct {
 	MaxQueueSize int64 `json:"max_queue_size"`
 	// Resource operations still queued when Drain ends, excluding the active upload.
 	UndrainedOperationCount int64 `json:"undrained_operation_count"`
-	// Failed update-building or upload attempts, excluding context cancellation.
+	// Failed update-building or upload attempts.
 	// Later successful updates do not clear the count.
 	RecordingErrorCount int64 `json:"recording_error_count"`
+	// Successful upload attempts. Repeated updates to a resource count separately.
+	RecordingSuccessCount int64 `json:"recording_success_count"`
 }
 
 // These metrics are experimental and are often added in an adhoc manner. There

From 699f3ec0e0d35300404315f48f2c22ab913baf9c Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 23:02:48 +0000
Subject: [PATCH 10/13] bundle: include total operation count in failed-upload
 telemetry output

---
 acceptance/bundle/dms/operation-upload-fails-once/output.txt | 1 +
 acceptance/bundle/dms/operation-upload-fails-once/script     | 2 +-
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index da8d71fec48..910a4f85964 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -71,6 +71,7 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
 === A failed history upload is counted separately from the queued operations
 >>> cat [OUTPUT_DIR]/out.requests.txt
 {
+  "operation_count": 1,
   "recording_error_count": 1,
   "recording_success_count": 1,
   "undrained_operation_count": 0
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script
index 977c7b04898..b1f37774bdf 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/script
+++ b/acceptance/bundle/dms/operation-upload-fails-once/script
@@ -24,7 +24,7 @@ trap '$CLI catalogs delete "$other_catalog" --force >> LOG.catalog 2>&1' EXIT
 trace update_file.py databricks.yml "catalog_name: main" "catalog_name: ${other_catalog}"
 trace $CLI bundle deploy --auto-approve
 title "A failed history upload is counted separately from the queued operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {recording_error_count, recording_success_count, undrained_operation_count}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, recording_error_count, recording_success_count, undrained_operation_count}'
 
 title "The service lists the schema with the failure"
 deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-upload-fails-once-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')

From fb1c5af59adb856c1dbdbc0ebea381159add37d3 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 23:17:43 +0000
Subject: [PATCH 11/13] bundle: report per-operation recording outcomes
 alongside upload attempts

---
 .../bundle/dms/multiple-resources/output.txt  |  3 ++
 .../bundle/dms/multiple-resources/script      |  2 +-
 .../operation-upload-fails-once/output.txt    |  3 ++
 .../dms/operation-upload-fails-once/script    |  2 +-
 bundle/phases/telemetry.go                    | 13 +++--
 libs/dms/operation_buffer.go                  | 54 ++++++++++++++++---
 libs/dms/operation_buffer_test.go             |  4 ++
 libs/telemetry/protos/bundle_deploy.go        |  6 +++
 8 files changed, 73 insertions(+), 14 deletions(-)

diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index 7d139818038..bad83b80484 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -14,6 +14,9 @@ Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
 >>> cat [OUTPUT_DIR]/out.requests.txt
 {
   "operation_count": 5,
+  "recorded_operation_count": 5,
+  "recording_failed_operation_count": 0,
+  "unrecorded_operation_count": 0,
   "undrained_operation_count": 0,
   "recording_error_count": 0,
   "recording_success_count": 5,
diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script
index 76c027bd559..49f572fadfa 100644
--- a/acceptance/bundle/dms/multiple-resources/script
+++ b/acceptance/bundle/dms/multiple-resources/script
@@ -3,7 +3,7 @@ envsubst < databricks.yml.tmpl > databricks.yml
 title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; other bundle/dms tests cover it."
 trace $CLI bundle deploy
 title "History telemetry survives version completion and counts all five staged operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, undrained_operation_count, recording_error_count, recording_success_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, recorded_operation_count, recording_failed_operation_count, unrecorded_operation_count, undrained_operation_count, recording_error_count, recording_success_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
 replace_ids.py
 trace print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index 910a4f85964..3383a04852f 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -72,6 +72,9 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
 >>> cat [OUTPUT_DIR]/out.requests.txt
 {
   "operation_count": 1,
+  "recorded_operation_count": 1,
+  "recording_failed_operation_count": 0,
+  "unrecorded_operation_count": 0,
   "recording_error_count": 1,
   "recording_success_count": 1,
   "undrained_operation_count": 0
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script
index b1f37774bdf..dc3bde4d669 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/script
+++ b/acceptance/bundle/dms/operation-upload-fails-once/script
@@ -24,7 +24,7 @@ trap '$CLI catalogs delete "$other_catalog" --force >> LOG.catalog 2>&1' EXIT
 trace update_file.py databricks.yml "catalog_name: main" "catalog_name: ${other_catalog}"
 trace $CLI bundle deploy --auto-approve
 title "A failed history upload is counted separately from the queued operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, recording_error_count, recording_success_count, undrained_operation_count}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, recorded_operation_count, recording_failed_operation_count, unrecorded_operation_count, recording_error_count, recording_success_count, undrained_operation_count}'
 
 title "The service lists the schema with the failure"
 deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-upload-fails-once-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go
index 1b1fa41e54a..8b91e489d35 100644
--- a/bundle/phases/telemetry.go
+++ b/bundle/phases/telemetry.go
@@ -407,11 +407,14 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 	var deploymentHistory *protos.BundleDeploymentHistory
 	if stats := b.DeploymentBundle.StateDB.OperationBufferStats(); stats != nil {
 		deploymentHistory = &protos.BundleDeploymentHistory{
-			OperationCount:          int64(stats.OperationCount),
-			MaxQueueSize:            int64(stats.MaxQueueSize),
-			UndrainedOperationCount: int64(stats.UndrainedOperationCount),
-			RecordingErrorCount:     int64(stats.RecordingErrorCount),
-			RecordingSuccessCount:   int64(stats.RecordingSuccessCount),
+			OperationCount:                int64(stats.OperationCount),
+			MaxQueueSize:                  int64(stats.MaxQueueSize),
+			UndrainedOperationCount:       int64(stats.UndrainedOperationCount),
+			RecordingErrorCount:           int64(stats.RecordingErrorCount),
+			RecordingSuccessCount:         int64(stats.RecordingSuccessCount),
+			RecordedOperationCount:        int64(stats.RecordedOperationCount),
+			RecordingFailedOperationCount: int64(stats.RecordingFailedOperationCount),
+			UnrecordedOperationCount:      int64(stats.UnrecordedOperationCount),
 		}
 	}
 
diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 9f93a6695d9..28a92534912 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -17,6 +17,14 @@ const bufferedOperations = 10
 
 const historyDrainTimeoutPerOperation = 5 * time.Second
 
+type recordingOutcome uint8
+
+const (
+	recordingPending recordingOutcome = iota
+	recordingSucceeded
+	recordingFailed
+)
+
 // BufferOptions selects optional deployment history shadow recording behavior.
 type BufferOptions struct {
 	BestEffort bool
@@ -27,11 +35,14 @@ type BufferOptions struct {
 
 // OperationBufferStats describes one deployment version's history uploads.
 type OperationBufferStats struct {
-	OperationCount          int
-	MaxQueueSize            int
-	UndrainedOperationCount int
-	RecordingErrorCount     int
-	RecordingSuccessCount   int
+	OperationCount                int
+	MaxQueueSize                  int
+	UndrainedOperationCount       int
+	RecordingErrorCount           int
+	RecordingSuccessCount         int
+	RecordedOperationCount        int
+	RecordingFailedOperationCount int
+	UnrecordedOperationCount      int
 }
 
 // OperationBuffer records each state write with DMS for one deployment version, off the apply
@@ -67,6 +78,8 @@ type OperationBuffer struct {
 	// pending holds the newest update per resource key, absent once the writer takes it.
 	pending map[string]OperationUpdate
 	stats   OperationBufferStats
+	// Missing staged keys are pending, including operations never queued.
+	outcomes map[string]recordingOutcome
 
 	// latestState is the newest state-bearing update per resource key, kept for the buffer's
 	// whole life - unlike pending, which the writer removes on take. RecordFailure merges its
@@ -101,6 +114,7 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		latestState:  make(map[string]OperationUpdate),
 		sequenceIDs:  make(map[string]int64),
 		stats:        OperationBufferStats{OperationCount: options.OperationCount},
+		outcomes:     make(map[string]recordingOutcome, options.OperationCount),
 	}
 	b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
 	go b.run(ctx)
@@ -114,6 +128,9 @@ func (b *OperationBuffer) RecordOperation(ctx context.Context, resourceKey strin
 	if err != nil {
 		b.mu.Lock()
 		b.stats.RecordingErrorCount++
+		if !inProgress {
+			b.outcomes[resourceKey] = recordingFailed
+		}
 		b.mu.Unlock()
 		b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		return
@@ -147,6 +164,7 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 		update = waiting.Merge(update)
 	}
 	b.pending[resourceKey] = update
+	delete(b.outcomes, resourceKey)
 	b.stats.MaxQueueSize = max(b.stats.MaxQueueSize, len(b.pending))
 	b.mu.Unlock()
 
@@ -214,11 +232,21 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio
 	})
 	b.mu.Lock()
 	defer b.mu.Unlock()
+	outcome := recordingSucceeded
 	if err != nil {
 		b.stats.RecordingErrorCount++
+		outcome = recordingFailed
+	} else {
+		b.stats.RecordingSuccessCount++
+	}
+	// A newer queued update or build failure takes precedence over this upload.
+	_, queued := b.pending[key]
+	if !queued && update.Status != StatusPending && b.outcomes[key] == recordingPending {
+		b.outcomes[key] = outcome
+	}
+	if err != nil {
 		return err
 	}
-	b.stats.RecordingSuccessCount++
 
 	// The next write for this resource echoes the sequence id this one earned.
 	b.sequenceIDs[key] = result.SequenceId
@@ -264,7 +292,19 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 func (b *OperationBuffer) Stats() OperationBufferStats {
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	return b.stats
+	stats := b.stats
+	for _, outcome := range b.outcomes {
+		switch outcome {
+		case recordingPending:
+			continue
+		case recordingSucceeded:
+			stats.RecordedOperationCount++
+		case recordingFailed:
+			stats.RecordingFailedOperationCount++
+		}
+	}
+	stats.UnrecordedOperationCount = stats.OperationCount - stats.RecordedOperationCount - stats.RecordingFailedOperationCount
+	return stats
 }
 
 // setErr keeps the first recording error for Drain to report.
diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go
index 0e019284360..506c9619b7e 100644
--- a/libs/dms/operation_buffer_test.go
+++ b/libs/dms/operation_buffer_test.go
@@ -27,6 +27,7 @@ func queued() *OperationBuffer {
 		queue:       make(chan string, bufferedOperations),
 		pending:     make(map[string]OperationUpdate),
 		latestState: make(map[string]OperationUpdate),
+		outcomes:    make(map[string]recordingOutcome),
 	}
 }
 
@@ -64,12 +65,15 @@ func TestBufferFailsOnOversizedState(t *testing.T) {
 	// Failing here says so, where reporting nothing would leave DMS without the resource
 	// and the next plan would create it again.
 	s := queued()
+	s.stats.OperationCount = 1
 
 	s.RecordOperation(t.Context(), "resources.jobs.foo", false, "id-1", json.RawMessage(strings.Repeat("x", maxStateSize+1)))
 
 	assert.ErrorContains(t, s.Err(), "exceeds the 65536 byte limit")
 	assert.Empty(t, s.queue)
 	assert.Empty(t, s.pending)
+	assert.Equal(t, 1, s.Stats().RecordingFailedOperationCount)
+	assert.Zero(t, s.Stats().UnrecordedOperationCount)
 }
 
 func TestDrainIsIdempotent(t *testing.T) {
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index bd8659ca37c..65919389083 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -60,6 +60,12 @@ type BundleDeploymentHistory struct {
 	RecordingErrorCount int64 `json:"recording_error_count"`
 	// Successful upload attempts. Repeated updates to a resource count separately.
 	RecordingSuccessCount int64 `json:"recording_success_count"`
+	// Resource operations whose final history update uploaded successfully.
+	RecordedOperationCount int64 `json:"recorded_operation_count"`
+	// Resource operations whose final history update failed to build or upload.
+	RecordingFailedOperationCount int64 `json:"recording_failed_operation_count"`
+	// Staged resource operations with no completed final history update.
+	UnrecordedOperationCount int64 `json:"unrecorded_operation_count"`
 }
 
 // These metrics are experimental and are often added in an adhoc manner. There

From 03655d4a00d4d23ce03cf061d4667eaa0da8ecc6 Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 23:31:14 +0000
Subject: [PATCH 12/13] bundle: use explicit operation, attempt, and queue
 telemetry names

---
 .../bundle/dms/multiple-resources/output.txt  | 14 +++---
 .../bundle/dms/multiple-resources/script      |  2 +-
 .../operation-upload-fails-once/output.txt    | 14 +++---
 .../dms/operation-upload-fails-once/script    |  2 +-
 bundle/direct/dstate/state.go                 |  4 +-
 bundle/phases/telemetry.go                    | 16 +++----
 libs/dms/operation_buffer.go                  | 44 +++++++++----------
 libs/dms/operation_buffer_test.go             |  8 ++--
 libs/telemetry/protos/bundle_deploy.go        | 16 +++----
 9 files changed, 60 insertions(+), 60 deletions(-)

diff --git a/acceptance/bundle/dms/multiple-resources/output.txt b/acceptance/bundle/dms/multiple-resources/output.txt
index bad83b80484..cdb857ee2ba 100644
--- a/acceptance/bundle/dms/multiple-resources/output.txt
+++ b/acceptance/bundle/dms/multiple-resources/output.txt
@@ -13,13 +13,13 @@ Resources: 5 created, 0 changed, 0 deleted, 0 unchanged
 === History telemetry survives version completion and counts all five staged operations
 >>> cat [OUTPUT_DIR]/out.requests.txt
 {
-  "operation_count": 5,
-  "recorded_operation_count": 5,
-  "recording_failed_operation_count": 0,
-  "unrecorded_operation_count": 0,
-  "undrained_operation_count": 0,
-  "recording_error_count": 0,
-  "recording_success_count": 5,
+  "staged_operation_count": 5,
+  "operation_recording_success_count": 5,
+  "operation_recording_failure_count": 0,
+  "operation_recording_pending_count": 0,
+  "queued_update_count_at_timeout": 0,
+  "recording_attempt_failure_count": 0,
+  "recording_attempt_success_count": 5,
   "queue_size_in_range": true
 }
 
diff --git a/acceptance/bundle/dms/multiple-resources/script b/acceptance/bundle/dms/multiple-resources/script
index 49f572fadfa..55e21494154 100644
--- a/acceptance/bundle/dms/multiple-resources/script
+++ b/acceptance/bundle/dms/multiple-resources/script
@@ -3,7 +3,7 @@ envsubst < databricks.yml.tmpl > databricks.yml
 title "Deploy several resources: operations are uploaded from background workers, so exactly one is recorded per resource no matter what order the uploads finish in. The serialized state is dropped from the output here; other bundle/dms tests cover it."
 trace $CLI bundle deploy
 title "History telemetry survives version completion and counts all five staged operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, recorded_operation_count, recording_failed_operation_count, unrecorded_operation_count, undrained_operation_count, recording_error_count, recording_success_count, queue_size_in_range: (.max_queue_size >= 1 and .max_queue_size <= 5)}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {staged_operation_count, operation_recording_success_count, operation_recording_failure_count, operation_recording_pending_count, queued_update_count_at_timeout, recording_attempt_failure_count, recording_attempt_success_count, queue_size_in_range: (.peak_pending_update_count >= 1 and .peak_pending_update_count <= 5)}'
 replace_ids.py
 trace print_requests.py --dms //versions/1/operations --sort --del-body state --oneline
 
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/output.txt b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
index 3383a04852f..72c48dfb96c 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/output.txt
+++ b/acceptance/bundle/dms/operation-upload-fails-once/output.txt
@@ -71,13 +71,13 @@ Resources: 1 created, 0 changed, 1 deleted, 0 unchanged
 === A failed history upload is counted separately from the queued operations
 >>> cat [OUTPUT_DIR]/out.requests.txt
 {
-  "operation_count": 1,
-  "recorded_operation_count": 1,
-  "recording_failed_operation_count": 0,
-  "unrecorded_operation_count": 0,
-  "recording_error_count": 1,
-  "recording_success_count": 1,
-  "undrained_operation_count": 0
+  "staged_operation_count": 1,
+  "operation_recording_success_count": 1,
+  "operation_recording_failure_count": 0,
+  "operation_recording_pending_count": 0,
+  "recording_attempt_failure_count": 1,
+  "recording_attempt_success_count": 1,
+  "queued_update_count_at_timeout": 0
 }
 
 === The service lists the schema with the failure
diff --git a/acceptance/bundle/dms/operation-upload-fails-once/script b/acceptance/bundle/dms/operation-upload-fails-once/script
index dc3bde4d669..94db48614ef 100644
--- a/acceptance/bundle/dms/operation-upload-fails-once/script
+++ b/acceptance/bundle/dms/operation-upload-fails-once/script
@@ -24,7 +24,7 @@ trap '$CLI catalogs delete "$other_catalog" --force >> LOG.catalog 2>&1' EXIT
 trace update_file.py databricks.yml "catalog_name: main" "catalog_name: ${other_catalog}"
 trace $CLI bundle deploy --auto-approve
 title "A failed history upload is counted separately from the queued operations"
-trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {operation_count, recorded_operation_count, recording_failed_operation_count, unrecorded_operation_count, recording_error_count, recording_success_count, undrained_operation_count}'
+trace cat "$OUT_REQUESTS" | jq 'select(.path? == "/telemetry-ext") | .body.protoLogs[] | fromjson | .entry.databricks_cli_log.bundle_deploy_event.deployment_history | {staged_operation_count, operation_recording_success_count, operation_recording_failure_count, operation_recording_pending_count, recording_attempt_failure_count, recording_attempt_success_count, queued_update_count_at_timeout}'
 
 title "The service lists the schema with the failure"
 deployment_id=$(MSYS_NO_PATHCONV=1 $CLI workspace get-status "/Workspace/Users/${CURRENT_USER_NAME}/.bundle/dms-upload-fails-once-${UNIQUE_NAME}/default/state/resources.deployment.json" | python3 -c 'import sys, json; print(json.load(sys.stdin)["object_id"])')
diff --git a/bundle/direct/dstate/state.go b/bundle/direct/dstate/state.go
index 138e9eb3519..326c3b8354c 100644
--- a/bundle/direct/dstate/state.go
+++ b/bundle/direct/dstate/state.go
@@ -185,8 +185,8 @@ func (db *DeploymentState) InitializeOperationBuffer(ctx context.Context, deploy
 	db.mu.Lock()
 	defer db.mu.Unlock()
 	options := dms.BufferOptions{
-		BestEffort:     db.mode == WritingDeploymentHistoryShadow,
-		OperationCount: operationCount,
+		BestEffort:           db.mode == WritingDeploymentHistoryShadow,
+		StagedOperationCount: operationCount,
 	}
 	db.operationBuffer = dms.StartOperationBuffer(ctx, deploymentID, versionID, options)
 	db.DeploymentID = deploymentID
diff --git a/bundle/phases/telemetry.go b/bundle/phases/telemetry.go
index 8b91e489d35..5fcd24db299 100644
--- a/bundle/phases/telemetry.go
+++ b/bundle/phases/telemetry.go
@@ -407,14 +407,14 @@ func LogDeployTelemetry(ctx context.Context, b *bundle.Bundle, errMsg string) {
 	var deploymentHistory *protos.BundleDeploymentHistory
 	if stats := b.DeploymentBundle.StateDB.OperationBufferStats(); stats != nil {
 		deploymentHistory = &protos.BundleDeploymentHistory{
-			OperationCount:                int64(stats.OperationCount),
-			MaxQueueSize:                  int64(stats.MaxQueueSize),
-			UndrainedOperationCount:       int64(stats.UndrainedOperationCount),
-			RecordingErrorCount:           int64(stats.RecordingErrorCount),
-			RecordingSuccessCount:         int64(stats.RecordingSuccessCount),
-			RecordedOperationCount:        int64(stats.RecordedOperationCount),
-			RecordingFailedOperationCount: int64(stats.RecordingFailedOperationCount),
-			UnrecordedOperationCount:      int64(stats.UnrecordedOperationCount),
+			StagedOperationCount:           int64(stats.StagedOperationCount),
+			PeakPendingUpdateCount:         int64(stats.PeakPendingUpdateCount),
+			QueuedUpdateCountAtTimeout:     int64(stats.QueuedUpdateCountAtTimeout),
+			RecordingAttemptFailureCount:   int64(stats.RecordingAttemptFailureCount),
+			RecordingAttemptSuccessCount:   int64(stats.RecordingAttemptSuccessCount),
+			OperationRecordingSuccessCount: int64(stats.OperationRecordingSuccessCount),
+			OperationRecordingFailureCount: int64(stats.OperationRecordingFailureCount),
+			OperationRecordingPendingCount: int64(stats.OperationRecordingPendingCount),
 		}
 	}
 
diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 28a92534912..75c4a0ef49e 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -28,21 +28,21 @@ const (
 // BufferOptions selects optional deployment history shadow recording behavior.
 type BufferOptions struct {
 	BestEffort bool
-	// OperationCount counts staged resource keys. All recorded keys must be staged.
+	// StagedOperationCount counts staged resource keys. All recorded keys must be staged.
 	// File history uses it for queue capacity and the drain timeout.
-	OperationCount int
+	StagedOperationCount int
 }
 
 // OperationBufferStats describes one deployment version's history uploads.
 type OperationBufferStats struct {
-	OperationCount                int
-	MaxQueueSize                  int
-	UndrainedOperationCount       int
-	RecordingErrorCount           int
-	RecordingSuccessCount         int
-	RecordedOperationCount        int
-	RecordingFailedOperationCount int
-	UnrecordedOperationCount      int
+	StagedOperationCount           int
+	PeakPendingUpdateCount         int
+	QueuedUpdateCountAtTimeout     int
+	RecordingAttemptFailureCount   int
+	RecordingAttemptSuccessCount   int
+	OperationRecordingSuccessCount int
+	OperationRecordingFailureCount int
+	OperationRecordingPendingCount int
 }
 
 // OperationBuffer records each state write with DMS for one deployment version, off the apply
@@ -100,7 +100,7 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		// Updates coalesce, so each staged key occupies at most one queue slot.
 		// The worker frees that slot before the key can be queued again.
 		// One slot per staged key therefore avoids deployment backpressure.
-		capacity = options.OperationCount
+		capacity = options.StagedOperationCount
 	}
 	b := &OperationBuffer{
 		deploymentID: deploymentID,
@@ -109,12 +109,12 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		done:         make(chan struct{}),
 		cancel:       cancel,
 		bestEffort:   options.BestEffort,
-		drainTimeout: time.Duration(options.OperationCount) * historyDrainTimeoutPerOperation,
+		drainTimeout: time.Duration(options.StagedOperationCount) * historyDrainTimeoutPerOperation,
 		pending:      make(map[string]OperationUpdate),
 		latestState:  make(map[string]OperationUpdate),
 		sequenceIDs:  make(map[string]int64),
-		stats:        OperationBufferStats{OperationCount: options.OperationCount},
-		outcomes:     make(map[string]recordingOutcome, options.OperationCount),
+		stats:        OperationBufferStats{StagedOperationCount: options.StagedOperationCount},
+		outcomes:     make(map[string]recordingOutcome, options.StagedOperationCount),
 	}
 	b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
 	go b.run(ctx)
@@ -127,7 +127,7 @@ func (b *OperationBuffer) RecordOperation(ctx context.Context, resourceKey strin
 	update, err := NewStateUpdate(resourceID, state, inProgress)
 	if err != nil {
 		b.mu.Lock()
-		b.stats.RecordingErrorCount++
+		b.stats.RecordingAttemptFailureCount++
 		if !inProgress {
 			b.outcomes[resourceKey] = recordingFailed
 		}
@@ -165,7 +165,7 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 	}
 	b.pending[resourceKey] = update
 	delete(b.outcomes, resourceKey)
-	b.stats.MaxQueueSize = max(b.stats.MaxQueueSize, len(b.pending))
+	b.stats.PeakPendingUpdateCount = max(b.stats.PeakPendingUpdateCount, len(b.pending))
 	b.mu.Unlock()
 
 	// when already queued: the writer reads the map when it gets to the key,
@@ -234,10 +234,10 @@ func (b *OperationBuffer) write(ctx context.Context, key string, update Operatio
 	defer b.mu.Unlock()
 	outcome := recordingSucceeded
 	if err != nil {
-		b.stats.RecordingErrorCount++
+		b.stats.RecordingAttemptFailureCount++
 		outcome = recordingFailed
 	} else {
-		b.stats.RecordingSuccessCount++
+		b.stats.RecordingAttemptSuccessCount++
 	}
 	// A newer queued update or build failure takes precedence over this upload.
 	_, queued := b.pending[key]
@@ -279,7 +279,7 @@ func (b *OperationBuffer) Drain(ctx context.Context) error {
 		// Deadline or cancellation: stop the upload so deployment can finish promptly.
 		b.mu.Lock()
 		defer b.mu.Unlock()
-		b.stats.UndrainedOperationCount = len(b.queue)
+		b.stats.QueuedUpdateCountAtTimeout = len(b.queue)
 		if b.err == nil {
 			b.err = ctx.Err()
 		}
@@ -298,12 +298,12 @@ func (b *OperationBuffer) Stats() OperationBufferStats {
 		case recordingPending:
 			continue
 		case recordingSucceeded:
-			stats.RecordedOperationCount++
+			stats.OperationRecordingSuccessCount++
 		case recordingFailed:
-			stats.RecordingFailedOperationCount++
+			stats.OperationRecordingFailureCount++
 		}
 	}
-	stats.UnrecordedOperationCount = stats.OperationCount - stats.RecordedOperationCount - stats.RecordingFailedOperationCount
+	stats.OperationRecordingPendingCount = stats.StagedOperationCount - stats.OperationRecordingSuccessCount - stats.OperationRecordingFailureCount
 	return stats
 }
 
diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go
index 506c9619b7e..101ae771bfc 100644
--- a/libs/dms/operation_buffer_test.go
+++ b/libs/dms/operation_buffer_test.go
@@ -65,15 +65,15 @@ func TestBufferFailsOnOversizedState(t *testing.T) {
 	// Failing here says so, where reporting nothing would leave DMS without the resource
 	// and the next plan would create it again.
 	s := queued()
-	s.stats.OperationCount = 1
+	s.stats.StagedOperationCount = 1
 
 	s.RecordOperation(t.Context(), "resources.jobs.foo", false, "id-1", json.RawMessage(strings.Repeat("x", maxStateSize+1)))
 
 	assert.ErrorContains(t, s.Err(), "exceeds the 65536 byte limit")
 	assert.Empty(t, s.queue)
 	assert.Empty(t, s.pending)
-	assert.Equal(t, 1, s.Stats().RecordingFailedOperationCount)
-	assert.Zero(t, s.Stats().UnrecordedOperationCount)
+	assert.Equal(t, 1, s.Stats().OperationRecordingFailureCount)
+	assert.Zero(t, s.Stats().OperationRecordingPendingCount)
 }
 
 func TestDrainIsIdempotent(t *testing.T) {
@@ -123,7 +123,7 @@ func TestHistoryQueueDoesNotBlockWhenWriterStops(t *testing.T) {
 		const operationCount = 1000
 		ctx, cancel := context.WithCancel(t.Context())
 		defer cancel()
-		b := StartOperationBuffer(ctx, "dep-1", 1, BufferOptions{BestEffort: true, OperationCount: operationCount})
+		b := StartOperationBuffer(ctx, "dep-1", 1, BufferOptions{BestEffort: true, StagedOperationCount: operationCount})
 		synctest.Wait()
 		cancel()
 		<-b.done
diff --git a/libs/telemetry/protos/bundle_deploy.go b/libs/telemetry/protos/bundle_deploy.go
index 65919389083..97deb98c7ac 100644
--- a/libs/telemetry/protos/bundle_deploy.go
+++ b/libs/telemetry/protos/bundle_deploy.go
@@ -50,22 +50,22 @@ type BundleDeployEvent struct {
 // BundleDeploymentHistory measures history recording for one deployment version.
 type BundleDeploymentHistory struct {
 	// Number of resource operations staged for this deployment version.
-	OperationCount int64 `json:"operation_count"`
+	StagedOperationCount int64 `json:"staged_operation_count"`
 	// Peak number of queued resource keys, excluding the in-flight upload.
-	MaxQueueSize int64 `json:"max_queue_size"`
+	PeakPendingUpdateCount int64 `json:"peak_pending_update_count"`
 	// Resource operations still queued when Drain ends, excluding the active upload.
-	UndrainedOperationCount int64 `json:"undrained_operation_count"`
+	QueuedUpdateCountAtTimeout int64 `json:"queued_update_count_at_timeout"`
 	// Failed update-building or upload attempts.
 	// Later successful updates do not clear the count.
-	RecordingErrorCount int64 `json:"recording_error_count"`
+	RecordingAttemptFailureCount int64 `json:"recording_attempt_failure_count"`
 	// Successful upload attempts. Repeated updates to a resource count separately.
-	RecordingSuccessCount int64 `json:"recording_success_count"`
+	RecordingAttemptSuccessCount int64 `json:"recording_attempt_success_count"`
 	// Resource operations whose final history update uploaded successfully.
-	RecordedOperationCount int64 `json:"recorded_operation_count"`
+	OperationRecordingSuccessCount int64 `json:"operation_recording_success_count"`
 	// Resource operations whose final history update failed to build or upload.
-	RecordingFailedOperationCount int64 `json:"recording_failed_operation_count"`
+	OperationRecordingFailureCount int64 `json:"operation_recording_failure_count"`
 	// Staged resource operations with no completed final history update.
-	UnrecordedOperationCount int64 `json:"unrecorded_operation_count"`
+	OperationRecordingPendingCount int64 `json:"operation_recording_pending_count"`
 }
 
 // These metrics are experimental and are often added in an adhoc manner. There

From 7af90d6cf7197e44f25f8e73ba7ab4c9f6dddbaa Mon Sep 17 00:00:00 2001
From: Shreyas Goenka <shreyas.goenka@databricks.com>
Date: Wed, 28 Oct 2026 23:53:43 +0000
Subject: [PATCH 13/13] bundle: attach recording state to each buffered update

---
 libs/dms/operation_buffer.go      | 62 +++++++++++++++++--------------
 libs/dms/operation_buffer_test.go | 31 +++++++++-------
 2 files changed, 51 insertions(+), 42 deletions(-)

diff --git a/libs/dms/operation_buffer.go b/libs/dms/operation_buffer.go
index 75c4a0ef49e..a74f24903dd 100644
--- a/libs/dms/operation_buffer.go
+++ b/libs/dms/operation_buffer.go
@@ -17,14 +17,19 @@ const bufferedOperations = 10
 
 const historyDrainTimeoutPerOperation = 5 * time.Second
 
-type recordingOutcome uint8
+type recordingState uint8
 
 const (
-	recordingPending recordingOutcome = iota
+	recordingPending recordingState = iota
 	recordingSucceeded
 	recordingFailed
 )
 
+type bufferedUpdate struct {
+	update OperationUpdate
+	state  recordingState
+}
+
 // BufferOptions selects optional deployment history shadow recording behavior.
 type BufferOptions struct {
 	BestEffort bool
@@ -76,10 +81,10 @@ type OperationBuffer struct {
 	mu sync.Mutex
 
 	// pending holds the newest update per resource key, absent once the writer takes it.
-	pending map[string]OperationUpdate
+	pending map[string]*bufferedUpdate
 	stats   OperationBufferStats
-	// Missing staged keys are pending, including operations never queued.
-	outcomes map[string]recordingOutcome
+	// Each operation points to its latest update; older uploads retain their own entry.
+	operations map[string]*bufferedUpdate
 
 	// latestState is the newest state-bearing update per resource key, kept for the buffer's
 	// whole life - unlike pending, which the writer removes on take. RecordFailure merges its
@@ -110,11 +115,11 @@ func StartOperationBuffer(ctx context.Context, deploymentID string, versionNum i
 		cancel:       cancel,
 		bestEffort:   options.BestEffort,
 		drainTimeout: time.Duration(options.StagedOperationCount) * historyDrainTimeoutPerOperation,
-		pending:      make(map[string]OperationUpdate),
+		pending:      make(map[string]*bufferedUpdate),
 		latestState:  make(map[string]OperationUpdate),
 		sequenceIDs:  make(map[string]int64),
 		stats:        OperationBufferStats{StagedOperationCount: options.StagedOperationCount},
-		outcomes:     make(map[string]recordingOutcome, options.StagedOperationCount),
+		operations:   make(map[string]*bufferedUpdate, options.StagedOperationCount),
 	}
 	b.stopQueue = sync.OnceFunc(func() { close(b.queue) })
 	go b.run(ctx)
@@ -129,7 +134,7 @@ func (b *OperationBuffer) RecordOperation(ctx context.Context, resourceKey strin
 		b.mu.Lock()
 		b.stats.RecordingAttemptFailureCount++
 		if !inProgress {
-			b.outcomes[resourceKey] = recordingFailed
+			b.operations[resourceKey] = &bufferedUpdate{state: recordingFailed}
 		}
 		b.mu.Unlock()
 		b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
@@ -161,10 +166,11 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 	}
 	waiting, queued := b.pending[resourceKey]
 	if queued {
-		update = waiting.Merge(update)
+		update = waiting.update.Merge(update)
 	}
-	b.pending[resourceKey] = update
-	delete(b.outcomes, resourceKey)
+	entry := &bufferedUpdate{update: update, state: recordingPending}
+	b.pending[resourceKey] = entry
+	b.operations[resourceKey] = entry
 	b.stats.PeakPendingUpdateCount = max(b.stats.PeakPendingUpdateCount, len(b.pending))
 	b.mu.Unlock()
 
@@ -178,13 +184,13 @@ func (b *OperationBuffer) record(resourceKey string, update OperationUpdate, kee
 }
 
 // take claims the update waiting for resourceKey.
-func (b *OperationBuffer) take(resourceKey string) (OperationUpdate, bool) {
+func (b *OperationBuffer) take(resourceKey string) (*bufferedUpdate, bool) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	update, ok := b.pending[resourceKey]
+	entry, ok := b.pending[resourceKey]
 	delete(b.pending, resourceKey)
-	return update, ok
+	return entry, ok
 }
 
 func (b *OperationBuffer) run(ctx context.Context) {
@@ -208,7 +214,7 @@ func (b *OperationBuffer) run(ctx context.Context) {
 		if !ok {
 			return
 		}
-		update, ok := b.take(resourceKey)
+		entry, ok := b.take(resourceKey)
 		if !ok {
 			// Unreachable: a key is queued only when nothing was waiting for it. Guard so
 			// a stray key could never write a zero-valued update.
@@ -216,34 +222,30 @@ func (b *OperationBuffer) run(ctx context.Context) {
 		}
 
 		// Keep going after a failure, so one bad write does not drop everything behind it.
-		if err := b.write(ctx, resourceKey, update); err != nil {
+		if err := b.write(ctx, resourceKey, entry); err != nil {
 			b.setErr(fmt.Errorf("recording operation for %s: %w", resourceKey, err))
 		}
 	}
 }
 
 // write sends one update, at the sequence id the resource is at.
-func (b *OperationBuffer) write(ctx context.Context, key string, update OperationUpdate) error {
+func (b *OperationBuffer) write(ctx context.Context, key string, entry *bufferedUpdate) error {
 	w := cmdctx.WorkspaceClient(ctx)
 	result, err := w.BundleDeployments.UpdateOperation(ctx, bundledeployments.UpdateOperationRequest{
 		Name:       OperationName(b.deploymentID, b.versionNum, key),
-		Operation:  update.operation(b.sequenceIDs[key]),
-		UpdateMask: update.Fields.Mask(),
+		Operation:  entry.update.operation(b.sequenceIDs[key]),
+		UpdateMask: entry.update.Fields.Mask(),
 	})
 	b.mu.Lock()
 	defer b.mu.Unlock()
-	outcome := recordingSucceeded
+	state := recordingSucceeded
 	if err != nil {
 		b.stats.RecordingAttemptFailureCount++
-		outcome = recordingFailed
+		state = recordingFailed
 	} else {
 		b.stats.RecordingAttemptSuccessCount++
 	}
-	// A newer queued update or build failure takes precedence over this upload.
-	_, queued := b.pending[key]
-	if !queued && update.Status != StatusPending && b.outcomes[key] == recordingPending {
-		b.outcomes[key] = outcome
-	}
+	entry.state = state
 	if err != nil {
 		return err
 	}
@@ -293,8 +295,12 @@ func (b *OperationBuffer) Stats() OperationBufferStats {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 	stats := b.stats
-	for _, outcome := range b.outcomes {
-		switch outcome {
+	for _, entry := range b.operations {
+		// A recreation's delete update is not the operation's final result.
+		if entry.update.Status == StatusPending {
+			continue
+		}
+		switch entry.state {
 		case recordingPending:
 			continue
 		case recordingSucceeded:
diff --git a/libs/dms/operation_buffer_test.go b/libs/dms/operation_buffer_test.go
index 101ae771bfc..6aa6aea4943 100644
--- a/libs/dms/operation_buffer_test.go
+++ b/libs/dms/operation_buffer_test.go
@@ -25,9 +25,9 @@ import (
 func queued() *OperationBuffer {
 	return &OperationBuffer{
 		queue:       make(chan string, bufferedOperations),
-		pending:     make(map[string]OperationUpdate),
+		pending:     make(map[string]*bufferedUpdate),
 		latestState: make(map[string]OperationUpdate),
-		outcomes:    make(map[string]recordingOutcome),
+		operations:  make(map[string]*bufferedUpdate),
 	}
 }
 
@@ -47,17 +47,20 @@ func TestBufferCoalescesWhileAKeyIsPending(t *testing.T) {
 	s.record("resources.jobs.foo", stateUpdate(t, "v2"), false, nil)
 
 	assert.Len(t, s.queue, 1)
-	update, ok := s.take("resources.jobs.foo")
+	entry, ok := s.take("resources.jobs.foo")
 	require.True(t, ok)
-	assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(update.State))
+	assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(entry.update.State))
+	first := entry
 
 	// Taken means a request has it, and an in-flight request cannot be recalled, so the next
 	// write gets its own slot rather than joining it.
 	s.record("resources.jobs.foo", stateUpdate(t, "v3"), false, nil)
 	assert.Len(t, s.queue, 2)
-	update, ok = s.take("resources.jobs.foo")
+	entry, ok = s.take("resources.jobs.foo")
 	require.True(t, ok)
-	assert.JSONEq(t, `{"state":{"name":"v3"}}`, string(update.State))
+	assert.JSONEq(t, `{"state":{"name":"v3"}}`, string(entry.update.State))
+	assert.NotSame(t, first, entry)
+	assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(first.update.State))
 }
 
 func TestBufferFailsOnOversizedState(t *testing.T) {
@@ -97,12 +100,12 @@ func TestRecordFailureKeepsTheStateTheLastWriteRecorded(t *testing.T) {
 	// state), which it must not use: the write already recorded v2, which the service may hold.
 	s.RecordFailure("resources.jobs.foo", "id-old", json.RawMessage(`{"state":{"name":"v1"}}`), errors.New("wait timed out"))
 
-	update, ok := s.take("resources.jobs.foo")
+	entry, ok := s.take("resources.jobs.foo")
 	require.True(t, ok)
-	assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(update.State), "keeps the state the write recorded")
-	assert.Equal(t, "id-new", update.ResourceID, "keeps the id the write recorded, not the pre-deploy id")
-	assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, update.Status)
-	assert.Contains(t, update.ErrorMessage, "wait timed out")
+	assert.JSONEq(t, `{"state":{"name":"v2"}}`, string(entry.update.State), "keeps the state the write recorded")
+	assert.Equal(t, "id-new", entry.update.ResourceID, "keeps the id the write recorded, not the pre-deploy id")
+	assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, entry.update.Status)
+	assert.Contains(t, entry.update.ErrorMessage, "wait timed out")
 }
 
 func TestRecordFailureUsesTheGivenStateWhenNoWriteRecorded(t *testing.T) {
@@ -112,10 +115,10 @@ func TestRecordFailureUsesTheGivenStateWhenNoWriteRecorded(t *testing.T) {
 
 	s.RecordFailure("resources.jobs.foo", "id-1", json.RawMessage(`{"state":{"name":"v1"}}`), errors.New("create failed"))
 
-	update, ok := s.take("resources.jobs.foo")
+	entry, ok := s.take("resources.jobs.foo")
 	require.True(t, ok)
-	assert.JSONEq(t, `{"state":{"name":"v1"}}`, string(update.State))
-	assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, update.Status)
+	assert.JSONEq(t, `{"state":{"name":"v1"}}`, string(entry.update.State))
+	assert.Equal(t, bundledeployments.OperationStatusOperationStatusFailed, entry.update.Status)
 }
 
 func TestHistoryQueueDoesNotBlockWhenWriterStops(t *testing.T) {