bundle: close the migration WAL before removing the temp state

bundle/direct/dstate/state.go

@@ -833,6 +833,26 @@ func (db *DeploymentState) Finalize(ctx context.Context) (resourcestate.Exported
 	return state, err
 }
 
+// Discard closes the state without persisting it and removes its file and WAL. It is for
+// temporary states, such as a migration's, that must not outlive a failure: the open WAL handle
+// is closed first, without which the removal fails on Windows.
+func (db *DeploymentState) Discard() {
+	db.mu.Lock()
+	path := db.Path
+	db.reset()
+	db.mu.Unlock()
+
+	if path != "" {
+		RemoveFiles(path)
+	}
+}
+
+// RemoveFiles removes the state file at path and its WAL. Missing files are ignored.
+func RemoveFiles(path string) {
+	_ = os.Remove(path)
+	_ = os.Remove(path + walSuffix)
+}
+
 // UpgradeToWrite transitions from read mode to write mode without re-reading state.
 // State must already be open for read. This initializes the WAL for writing.
 func (db *DeploymentState) UpgradeToWrite() error {

bundle/direct/dstate/state_test.go

@@ -45,6 +45,27 @@ func TestFinalizeWithNoEntriesDoesNotWriteStateFile(t *testing.T) {
 	assert.ErrorIs(t, err, os.ErrNotExist)
 }
 
+func TestDiscardRemovesStateAndWAL(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "state.json")
+
+	var db DeploymentState
+	db.OpenWithData(path, NewDatabase("lineage", 1))
+	require.NoError(t, db.UpgradeToWrite())
+	require.NoError(t, db.SaveState(t.Context(), "jobs.my_job", "123", map[string]string{"key": "val"}, nil))
+	db.Discard()
+
+	_, err := os.Stat(path)
+	assert.ErrorIs(t, err, os.ErrNotExist)
+	_, err = os.Stat(path + walSuffix)
+	assert.ErrorIs(t, err, os.ErrNotExist)
+
+	// The WAL is opened with O_EXCL, so a leftover one would fail a fresh write at the same path.
+	var db2 DeploymentState
+	db2.OpenWithData(path, NewDatabase("lineage", 1))
+	require.NoError(t, db2.UpgradeToWrite())
+	db2.Discard()
+}
+
 func TestExportStateFromDataJobRunJobID(t *testing.T) {
 	data := Database{
 		State: map[string]ResourceEntry{

bundle/statemgmt/direct_migration.go

@@ -67,10 +67,7 @@ func Migrate(ctx context.Context, b *bundle.Bundle) (bool, error) {
 	if tempStatePath != "" {
 		// Always remove the temp state and its WAL. The converted state is loaded into memory
 		// below and never renamed into place, so these are the only on-disk copies to clean up.
-		defer func() {
-			_ = os.Remove(tempStatePath)
-			_ = os.Remove(tempStatePath + ".wal")
-		}()
+		defer dstate.RemoveFiles(tempStatePath)
 	}
 	if hasWarnings {
 		b.Metrics.SetBoolValue(metrics.DirectMigrateWarnings, true)
@@ -233,8 +230,7 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 	tempStatePath := filepath.Join(filepath.Dir(localDirectPath), "resources.migrating.json")
 	// Clean up any leftovers from a crashed previous run so UpgradeToWrite
 	// (which opens the .wal with O_EXCL) succeeds.
-	_ = os.Remove(tempStatePath)
-	_ = os.Remove(tempStatePath + ".wal")
+	dstate.RemoveFiles(tempStatePath)
 
 	// SecretScopeFixups and the direct-engine state builder report failures via
 	// logdiag. Run them in an isolated + collecting context so their diagnostics
@@ -258,6 +254,9 @@ func convertTFStateToDirect(ctx context.Context, b *bundle.Bundle, tfState *migr
 	// stays at tf+1. Either way it outranks the terraform state, which is all engine selection needs.
 	var stateDB dstate.DeploymentState
 	stateDB.OpenWithData(tempStatePath, dstate.NewDatabase(tfState.Lineage, tfState.Serial+1))
+	// On a failed conversion, close the WAL so the caller can remove the files (Windows refuses to
+	// remove an open file). After the successful Finalize below this is a no-op.
+	defer stateDB.Discard()
 
 	// Apply SecretScopeFixups so the config matches what the direct engine expects.
 	// This adds MANAGE ACL for the current user to all secret scopes, ensuring