diff --git a/README.md b/README.md
index 0476693..266f1b5 100644
--- a/README.md
+++ b/README.md
@@ -346,6 +346,8 @@ elasticsearch:
     indicesPattern: sts*,.ds-sts_k8s_logs*
 ```
 
+`indicesPattern` selects indices to delete and restore. The restore names the `datastreamName` data stream, which brings back all of its snapshotted backing indices, so the pattern must select all of the logs backing indices or none of them; a pattern that selects only some is rejected before anything is scaled down or deleted.
+
 Apply to Kubernetes:
 
 ```bash
diff --git a/cmd/elasticsearch/check_and_finalize.go b/cmd/elasticsearch/check_and_finalize.go
index c6b6197..f605254 100644
--- a/cmd/elasticsearch/check_and_finalize.go
+++ b/cmd/elasticsearch/check_and_finalize.go
@@ -2,6 +2,7 @@ package elasticsearch
 
 import (
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"time"
@@ -104,7 +105,7 @@ const (
 // health check. It starts from the caller's client so the happy path opens no extra port-forward.
 type reconnectingHealthClient struct {
 	appCtx *app.Context
-	client indicesHealthGetter
+	client es.Interface
 	pf     *portforward.Conn
 }
 
@@ -126,6 +127,40 @@ func (r *reconnectingHealthClient) GetIndicesHealth() (map[string]es.IndexHealth
 	return health, nil
 }
 
+// GetDatastream and ListIndices let the post-restore verification survive a tunnel that dropped
+// during a long restore, the same way polling does.
+func (r *reconnectingHealthClient) GetDatastream(name string) ([]string, bool, error) {
+	if r.client == nil {
+		if err := r.connect(); err != nil {
+			return nil, false, err
+		}
+	}
+
+	indices, exists, err := r.client.GetDatastream(name)
+	if err != nil {
+		r.disconnect()
+		return nil, false, err
+	}
+
+	return indices, exists, nil
+}
+
+func (r *reconnectingHealthClient) ListIndices(pattern string) ([]string, error) {
+	if r.client == nil {
+		if err := r.connect(); err != nil {
+			return nil, err
+		}
+	}
+
+	indices, err := r.client.ListIndices(pattern)
+	if err != nil {
+		r.disconnect()
+		return nil, err
+	}
+
+	return indices, nil
+}
+
 func (r *reconnectingHealthClient) connect() error {
 	pf, err := portforward.SetupPortForward(
 		r.appCtx.K8sClient,
@@ -427,7 +462,7 @@ func checkAndFinalize(
 		switch status {
 		case es.StatusSuccess:
 			appCtx.Logger.Successf("Restore completed successfully")
-			return finalizeRestore(appCtx)
+			return verifyAndFinalize(healthClient, appCtx, expected)
 		default:
 			return fmt.Errorf("restore completed with unexpected status: %s", status)
 		}
@@ -438,7 +473,9 @@ func checkAndFinalize(
 
 	if waitForComplete {
 		appCtx.Logger.Println()
-		return waitAndFinalize(statusFn, appCtx, snapshotName)
+		return waitAndFinalize(statusFn, appCtx, snapshotName, func() error {
+			return verifyAndFinalize(healthClient, appCtx, expected)
+		})
 	}
 
 	// Not waiting - print status and exit
@@ -448,17 +485,96 @@ func checkAndFinalize(
 }
 
 // waitAndFinalize waits for restore to complete and finalizes (scale up)
-func waitAndFinalize(statusFn func() (string, bool, error), appCtx *app.Context, snapshotName string) error {
+func waitAndFinalize(statusFn func() (string, bool, error), appCtx *app.Context, snapshotName string, finalize func() error) error {
 	restore.PrintAPIWaitingMessage("elasticsearch", snapshotName, appCtx.Namespace, appCtx.Logger)
 
 	if err := restore.WaitForAPIRestore(statusFn, 0, appCtx.Logger); err != nil {
 		return err
 	}
 
-	// Finalize restore (scale up)
+	return finalize()
+}
+
+// logsDatastreamReader is the part of the Elasticsearch client the post-restore verification needs.
+type logsDatastreamReader interface {
+	GetDatastream(datastreamName string) ([]string, bool, error)
+	ListIndices(pattern string) ([]string, error)
+}
+
+const verifyAttempts = 3
+
+// verifyRetryInterval is a variable so tests do not wait.
+var verifyRetryInterval = 2 * time.Second
+
+// withRetries retries transient request failures, such as a port-forward that dropped. It does not
+// apply to verification verdicts, which are returned by the caller after the data is fetched.
+func withRetries(log *logger.Logger, what string, fn func() error) error {
+	var err error
+	for attempt := 1; attempt <= verifyAttempts; attempt++ {
+		if err = fn(); err == nil {
+			return nil
+		}
+		if attempt < verifyAttempts {
+			log.Warningf("%s failed (%d/%d), retrying: %v", what, attempt, verifyAttempts, err)
+			time.Sleep(verifyRetryInterval)
+		}
+	}
+
+	return err
+}
+
+// verifyAndFinalize refuses to scale up over a restore that left the logs outside the data stream.
+func verifyAndFinalize(client logsDatastreamReader, appCtx *app.Context, expected []string) error {
+	if err := verifyLogsDatastream(client, appCtx.Logger, appCtx.Config.Elasticsearch.Restore, expected); err != nil {
+		return withFinalizeHint(err, appCtx.Namespace)
+	}
+
 	return finalizeRestore(appCtx)
 }
 
+// verifyLogsDatastream checks that the logs data stream exists and that no logs backing index sits
+// outside it. It does nothing when the restore did not include logs backing indices.
+func verifyLogsDatastream(client logsDatastreamReader, log *logger.Logger, restoreConfig config.RestoreConfig, expected []string) error {
+	if !hasDatastreamIndices(expected, restoreConfig.DatastreamIndexPrefix) {
+		return nil
+	}
+
+	var backing []string
+	var exists bool
+	err := withRetries(log, "Data stream check", func() (err error) {
+		backing, exists, err = client.GetDatastream(restoreConfig.DatastreamName)
+		return err
+	})
+	if err != nil {
+		return fmt.Errorf("failed to verify datastream '%s': %w", restoreConfig.DatastreamName, err)
+	}
+	if !exists {
+		return fmt.Errorf("restore completed but the '%s' data stream does not exist", restoreConfig.DatastreamName)
+	}
+
+	var allIndices []string
+	err = withRetries(log, "Index listing", func() (err error) {
+		allIndices, err = client.ListIndices(restoreConfig.DatastreamIndexPrefix + "-*")
+		return err
+	})
+	if err != nil {
+		return fmt.Errorf("failed to list indices: %w", err)
+	}
+	var orphans []string
+	for _, index := range allIndices {
+		if isDatastreamBackingIndex(index, restoreConfig.DatastreamIndexPrefix) && !slices.Contains(backing, index) {
+			orphans = append(orphans, index)
+		}
+	}
+	if len(orphans) > 0 {
+		sort.Strings(orphans)
+		return fmt.Errorf("restore completed but %d '%s' backing index(es) are outside the data stream: %s",
+			len(orphans), restoreConfig.DatastreamName, strings.Join(orphans, ", "))
+	}
+
+	return nil
+}
+
 // finalizeRestore performs post-restore finalization (scale up deployments and release lock)
 func finalizeRestore(appCtx *app.Context) error {
 	appCtx.Logger.Println()
diff --git a/cmd/elasticsearch/check_and_finalize_test.go b/cmd/elasticsearch/check_and_finalize_test.go
index ad1c185..74aea1f 100644
--- a/cmd/elasticsearch/check_and_finalize_test.go
+++ b/cmd/elasticsearch/check_and_finalize_test.go
@@ -22,6 +22,7 @@ type healthResult struct {
 
 // fakeHealthClient replays results in order, repeating the last one once exhausted.
 type fakeHealthClient struct {
+	es.Interface
 	results []healthResult
 	calls   int
 }
@@ -541,6 +542,20 @@ func TestReconnectingHealthClient(t *testing.T) {
 		assert.Nil(t, client.pf, "must not adopt a port-forward it did not open")
 	})
 
+	t.Run("verification calls drop a broken client so the next call reconnects", func(t *testing.T) {
+		broken := &brokenDatastreamClient{}
+		client := &reconnectingHealthClient{client: broken}
+
+		_, _, err := client.GetDatastream("sts_k8s_logs")
+		require.Error(t, err)
+		assert.Nil(t, client.client)
+
+		client.client = broken
+		_, err = client.ListIndices("*")
+		require.Error(t, err)
+		assert.Nil(t, client.client)
+	})
+
 	t.Run("disconnect is safe to call repeatedly", func(t *testing.T) {
 		client := &reconnectingHealthClient{client: &fakeHealthClient{}}
 
@@ -550,3 +565,15 @@ func TestReconnectingHealthClient(t *testing.T) {
 		})
 	})
 }
+
+type brokenDatastreamClient struct {
+	es.Interface
+}
+
+func (brokenDatastreamClient) GetDatastream(string) ([]string, bool, error) {
+	return nil, false, errors.New("connection refused")
+}
+
+func (brokenDatastreamClient) ListIndices(string) ([]string, error) {
+	return nil, errors.New("connection refused")
+}
diff --git a/cmd/elasticsearch/configure_test.go b/cmd/elasticsearch/configure_test.go
index 396209c..c24cbff 100644
--- a/cmd/elasticsearch/configure_test.go
+++ b/cmd/elasticsearch/configure_test.go
@@ -96,8 +96,12 @@ func (m *mockESClientForConfigure) RestoreSnapshot(_, _, _ string, _ bool) error
 	return fmt.Errorf("not implemented")
 }
 
-func (m *mockESClientForConfigure) RolloverDatastream(_ string) error {
-	return fmt.Errorf("not implemented")
+func (m *mockESClientForConfigure) DeleteDatastream(_ string) (bool, error) {
+	return false, fmt.Errorf("not implemented")
+}
+
+func (m *mockESClientForConfigure) GetDatastream(_ string) ([]string, bool, error) {
+	return nil, false, fmt.Errorf("not implemented")
 }
 
 // TestConfigureCmd_Unit tests the command structure
diff --git a/cmd/elasticsearch/list_test.go b/cmd/elasticsearch/list_test.go
index 3a7b45e..9016cc5 100644
--- a/cmd/elasticsearch/list_test.go
+++ b/cmd/elasticsearch/list_test.go
@@ -273,8 +273,12 @@ func (m *mockESClient) ConfigureSLMPolicy(_, _, _, _, _, _ string, _, _ int) err
 	return fmt.Errorf("not implemented")
 }
 
-func (m *mockESClient) RolloverDatastream(_ string) error {
-	return fmt.Errorf("not implemented")
+func (m *mockESClient) DeleteDatastream(_ string) (bool, error) {
+	return false, fmt.Errorf("not implemented")
+}
+
+func (m *mockESClient) GetDatastream(_ string) ([]string, bool, error) {
+	return nil, false, fmt.Errorf("not implemented")
 }
 
 // createFakeClientWithConfig creates a fake Kubernetes client with a ConfigMap containing the given config
diff --git a/cmd/elasticsearch/restore.go b/cmd/elasticsearch/restore.go
index 4c25f98..f4ec0e0 100644
--- a/cmd/elasticsearch/restore.go
+++ b/cmd/elasticsearch/restore.go
@@ -3,6 +3,7 @@ package elasticsearch
 import (
 	"fmt"
 	"path"
+	"slices"
 	"sort"
 	"strings"
 	"time"
@@ -39,7 +40,7 @@ func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "restore",
 		Short: "Restore Elasticsearch from a snapshot",
-		Long:  `Restore Elasticsearch indices from a snapshot. Deletes existing STS indices before restore. Waits for completion by default; use --background to run asynchronously.`,
+		Long:  `Restore Elasticsearch indices and the logs data stream from a snapshot. Deletes existing STS indices and the logs data stream before restore. Waits for completion by default; use --background to run asynchronously.`,
 		Run: func(_ *cobra.Command, _ []string) {
 			cmdutils.Run(globalFlags, runRestore, cmdutils.StorageIsRequired)
 		}}
@@ -99,7 +100,7 @@ func runRestore(appCtx *app.Context) error {
 
 	// Reject snapshots that cannot be monitored safely before scaling down workloads or deleting
 	// any indices. check-and-finalize repeats this validation for independently resumed operations.
-	restoreIndices, err := restorableSnapshotIndices(snapshotDetails, appCtx.Config.Elasticsearch.Restore)
+	targets, err := prepareRestoreTargets(snapshotDetails, appCtx.Config.Elasticsearch.Restore)
 	if err != nil {
 		return err
 	}
@@ -135,27 +136,53 @@ func runRestore(appCtx *app.Context) error {
 		return err
 	}
 
-	// Delete all STS indices before restore
+	if err := replaceSTSIndices(esClient, appCtx, repository, snapshotDetails, targets); err != nil {
+		return withFinalizeHint(err, appCtx.Namespace)
+	}
+
+	if runBackground {
+		restore.PrintAPIRunningRestoreStatus("elasticsearch", selectedSnapshot, appCtx.Namespace, appCtx.Logger)
+		return nil
+	}
+
+	return checkAndFinalize(esClient, appCtx, repository, selectedSnapshot, !runBackground, restoreNoProgressIn)
+}
+
+// prepareRestoreTargets validates the snapshot and builds the restore request. It needs no cluster
+// access, so it runs before anything is scaled down or deleted.
+func prepareRestoreTargets(snapshot *es.Snapshot, restoreConfig config.RestoreConfig) ([]string, error) {
+	restoreIndices, err := restorableSnapshotIndices(snapshot, restoreConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return restoreTargets(snapshot, restoreIndices, restoreConfig)
+}
+
+// withFinalizeHint tells the user how to release workloads that stay scaled down after a failure.
+func withFinalizeHint(err error, namespace string) error {
+	return fmt.Errorf("%w. Workloads are still scaled down; after fixing this, release them with: "+
+		"sts-backup elasticsearch check-and-finalize --finalize-only -n %s", err, namespace)
+}
+
+// replaceSTSIndices deletes the target's STS indices and triggers an asynchronous restore of the snapshot's.
+func replaceSTSIndices(esClient es.Interface, appCtx *app.Context, repository string, snapshot *es.Snapshot, targets []string) error {
 	appCtx.Logger.Println()
 	if err := deleteAllSTSIndices(esClient, appCtx); err != nil {
 		return err
 	}
 
-	// Trigger async restore
 	appCtx.Logger.Println()
-	isPartial := snapshotDetails.State == "PARTIAL"
-	appCtx.Logger.Infof("Triggering restore for snapshot: %s", selectedSnapshot)
-	if err := esClient.RestoreSnapshot(repository, selectedSnapshot, strings.Join(restoreIndices, ","), isPartial); err != nil {
-		return fmt.Errorf("failed to trigger restore: %w", err)
+	isPartial := snapshot.State == "PARTIAL"
+	appCtx.Logger.Infof("Triggering restore for snapshot: %s", snapshot.Snapshot)
+	if err := esClient.RestoreSnapshot(repository, snapshot.Snapshot, strings.Join(targets, ","), isPartial); err != nil {
+		return fmt.Errorf("failed to trigger restore (if the error names an existing index or data stream, "+
+			"a workload not matched by scaleDownLabelSelector %q may be writing to Elasticsearch): %w",
+			appCtx.Config.Elasticsearch.Restore.ScaleDownLabelSelector, err)
 	}
 	appCtx.Logger.Successf("Restore triggered successfully")
 
-	if runBackground {
-		restore.PrintAPIRunningRestoreStatus("elasticsearch", selectedSnapshot, appCtx.Namespace, appCtx.Logger)
-		return nil
-	}
-
-	return checkAndFinalize(esClient, appCtx, repository, selectedSnapshot, !runBackground, restoreNoProgressIn)
+	return nil
 }
 
 // getLatestSnapshot retrieves the most recent snapshot from the repository
@@ -210,50 +237,113 @@ func validateSnapshotState(snapshot *es.Snapshot, appCtx *app.Context, skipConfi
 	}
 }
 
-// deleteAllSTSIndices deletes all STS indices including datastream rollover if needed
+// deleteAllSTSIndices deletes all STS indices, including the logs data stream and any of its
+// backing indices left without one.
 func deleteAllSTSIndices(esClient es.Interface, appCtx *app.Context) error {
-	appCtx.Logger.Infof("Fetching current Elasticsearch indices...")
-	allIndices, err := esClient.ListIndices("*")
-	if err != nil {
-		return fmt.Errorf("failed to list indices: %w", err)
-	}
+	restoreConfig := appCtx.Config.Elasticsearch.Restore
 
-	stsIndices := filterSTSIndices(allIndices, appCtx.Config.Elasticsearch.Restore.IndexPrefix, appCtx.Config.Elasticsearch.Restore.DatastreamIndexPrefix)
-	stsIndices, err = filterIndicesByPattern(stsIndices, appCtx.Config.Elasticsearch.Restore.IndicesPattern)
+	appCtx.Logger.Infof("Fetching current Elasticsearch indices...")
+	stsIndices, err := listSTSIndices(esClient, restoreConfig)
 	if err != nil {
-		return fmt.Errorf("invalid Elasticsearch restore indicesPattern: %w", err)
+		return err
 	}
 
 	if len(stsIndices) == 0 {
 		appCtx.Logger.Infof("No STS indices found to delete")
-		return nil
-	}
-
-	appCtx.Logger.Infof("Found %d STS index(es) to delete", len(stsIndices))
-	for _, index := range stsIndices {
-		appCtx.Logger.Debugf("  - %s", index)
+	} else {
+		appCtx.Logger.Infof("Found %d STS index(es) to delete", len(stsIndices))
+		for _, index := range stsIndices {
+			appCtx.Logger.Debugf("  - %s", index)
+		}
 	}
 
-	// Check for datastream and rollover if needed
-	if hasDatastreamIndices(stsIndices, appCtx.Config.Elasticsearch.Restore.DatastreamIndexPrefix) {
-		appCtx.Logger.Infof("Rolling over datastream '%s'...", appCtx.Config.Elasticsearch.Restore.DatastreamName)
-		if err := esClient.RolloverDatastream(appCtx.Config.Elasticsearch.Restore.DatastreamName); err != nil {
-			return fmt.Errorf("failed to rollover datastream: %w", err)
+	// Rolling over instead would leave a new write index behind that can collide with the snapshot.
+	if hasDatastreamIndices(stsIndices, restoreConfig.DatastreamIndexPrefix) {
+		appCtx.Logger.Infof("Deleting datastream '%s'...", restoreConfig.DatastreamName)
+		deleted, err := esClient.DeleteDatastream(restoreConfig.DatastreamName)
+		if err != nil {
+			return fmt.Errorf("failed to delete datastream: %w", err)
+		}
+		if deleted {
+			appCtx.Logger.Successf("Datastream deleted successfully")
+			if stsIndices, err = listSTSIndices(esClient, restoreConfig); err != nil {
+				return err
+			}
+		} else {
+			appCtx.Logger.Infof("Datastream '%s' does not exist; deleting its backing indices directly", restoreConfig.DatastreamName)
 		}
-		appCtx.Logger.Successf("Datastream rolled over successfully")
 	}
 
-	// Delete all indices
-	appCtx.Logger.Infof("Deleting %d index(es)...", len(stsIndices))
+	if len(stsIndices) > 0 {
+		appCtx.Logger.Infof("Deleting %d index(es)...", len(stsIndices))
+	}
 	for _, index := range stsIndices {
 		if err := deleteIndexWithVerification(esClient, index, appCtx.Logger); err != nil {
 			return err
 		}
 	}
+
+	// A workload that is still writing recreates indices, which would restore into a dirty target.
+	remaining, err := listSTSIndices(esClient, restoreConfig)
+	if err != nil {
+		return err
+	}
+	if len(remaining) > 0 {
+		return fmt.Errorf("STS indices reappeared after deletion, so a workload is still writing to Elasticsearch: %s. "+
+			"Check for writers not matched by scaleDownLabelSelector %q",
+			strings.Join(remaining, ", "), restoreConfig.ScaleDownLabelSelector)
+	}
 	appCtx.Logger.Successf("All indices deleted successfully")
 	return nil
 }
 
+func listSTSIndices(esClient es.Interface, restoreConfig config.RestoreConfig) ([]string, error) {
+	allIndices, err := esClient.ListIndices("*")
+	if err != nil {
+		return nil, fmt.Errorf("failed to list indices: %w", err)
+	}
+
+	stsIndices := filterSTSIndices(allIndices, restoreConfig.IndexPrefix, restoreConfig.DatastreamIndexPrefix)
+	stsIndices, err = filterIndicesByPattern(stsIndices, restoreConfig.IndicesPattern)
+	if err != nil {
+		return nil, fmt.Errorf("invalid Elasticsearch restore indicesPattern: %w", err)
+	}
+
+	return stsIndices, nil
+}
+
+// restoreTargets adds the logs data stream to the restore request. Naming only its backing indices
+// restores them as plain indices and leaves the data stream missing. Naming a data stream the
+// snapshot lacks fails the whole restore, so that case is rejected up front.
+func restoreTargets(snapshot *es.Snapshot, indices []string, restoreConfig config.RestoreConfig) ([]string, error) {
+	if !hasDatastreamIndices(indices, restoreConfig.DatastreamIndexPrefix) {
+		return indices, nil
+	}
+
+	if !slices.Contains(snapshot.DataStreams, restoreConfig.DatastreamName) {
+		return nil, fmt.Errorf("snapshot %s contains %s backing indices but not the %s data stream, "+
+			"so the logs cannot be restored as a data stream; use a snapshot taken by the backup SLM policy",
+			snapshot.Snapshot, restoreConfig.DatastreamIndexPrefix, restoreConfig.DatastreamName)
+	}
+
+	// Naming the data stream restores all of its snapshotted backing indices, which completion
+	// tracking and the deletion would not cover if the pattern selected only some of them.
+	var omitted []string
+	for _, index := range snapshot.Indices {
+		if isDatastreamBackingIndex(index, restoreConfig.DatastreamIndexPrefix) && !slices.Contains(indices, index) {
+			omitted = append(omitted, index)
+		}
+	}
+	if len(omitted) > 0 {
+		slices.Sort(omitted)
+		return nil, fmt.Errorf("indicesPattern %q selects only some %s backing indices, but restoring the %s data stream "+
+			"restores all of them; omitted: %s. The pattern must select all of them or none",
+			restoreConfig.IndicesPattern, restoreConfig.DatastreamIndexPrefix, restoreConfig.DatastreamName, strings.Join(omitted, ", "))
+	}
+
+	return append(slices.Clone(indices), restoreConfig.DatastreamName), nil
+}
+
 // filterSTSIndices filters indices that match the configured STS prefixes
 func filterSTSIndices(allIndices []string, indexPrefix, datastreamPrefix string) []string {
 	var stsIndices []string
diff --git a/cmd/elasticsearch/restore_test.go b/cmd/elasticsearch/restore_test.go
index 33531e5..e084f12 100644
--- a/cmd/elasticsearch/restore_test.go
+++ b/cmd/elasticsearch/restore_test.go
@@ -2,6 +2,8 @@ package elasticsearch
 
 import (
 	"fmt"
+	"slices"
+	"strings"
 	"testing"
 	"time"
 
@@ -21,13 +23,47 @@ type mockESClientForRestore struct {
 	indexExistsMap   map[string]bool
 	restoreErr       error
 	getSnapshotErr   error
-	rolloverErr      error
 	deletedIndices   []string
 	restoredSnapshot string
-	rolledOverDS     string
+	restoredTargets  string
+	restoredPartial  bool
+	// datastreams maps a live data stream to its backing indices.
+	datastreams           map[string][]string
+	datastreamDeleteCalls []string
+	deleteDatastreamErr   error
+	// listCalls counts ListIndices calls; addOnListCall adds indices on the given call, modelling a writer.
+	listCalls     int
+	addOnListCall map[int][]string
+	// restoredDatastreams and restoredPlain record what _restore brought back: a named data stream
+	// gets its backing indices, anything else comes back as plain indices.
+	restoredDatastreams map[string][]string
+	restoredPlain       []string
+}
+
+// settle makes the indices brought back by _restore visible to ListIndices.
+func (m *mockESClientForRestore) settle() {
+	m.indices = slices.Clone(m.restoredPlain)
+	for _, backing := range m.restoredDatastreams {
+		m.indices = append(m.indices, backing...)
+	}
+}
+
+func (m *mockESClientForRestore) GetDatastream(name string) ([]string, bool, error) {
+	backing, ok := m.restoredDatastreams[name]
+	return backing, ok, nil
 }
 
 func (m *mockESClientForRestore) ListIndices(_ string) ([]string, error) {
+	m.listCalls++
+	if added := m.addOnListCall[m.listCalls]; len(added) > 0 {
+		if m.indexExistsMap == nil {
+			m.indexExistsMap = make(map[string]bool)
+		}
+		for _, index := range added {
+			m.indices = append(m.indices, index)
+			m.indexExistsMap[index] = true
+		}
+	}
 	return m.indices, nil
 }
 
@@ -42,7 +78,13 @@ func (m *mockESClientForRestore) DeleteIndex(index string) error {
 	if m.deleteErr != nil {
 		return m.deleteErr
 	}
+	for _, backing := range m.datastreams {
+		if len(backing) > 0 && backing[len(backing)-1] == index {
+			return fmt.Errorf("index [%s] is the write index for data stream and cannot be deleted", index)
+		}
+	}
 	m.deletedIndices = append(m.deletedIndices, index)
+	m.indices = slices.DeleteFunc(m.indices, func(i string) bool { return i == index })
 	// Update exists map
 	if m.indexExistsMap == nil {
 		m.indexExistsMap = make(map[string]bool)
@@ -62,20 +104,66 @@ func (m *mockESClientForRestore) IndexExists(index string) (bool, error) {
 	return exists, nil
 }
 
-func (m *mockESClientForRestore) RestoreSnapshot(_, snapshotName, _ string, _ bool) error {
+// RestoreSnapshot fails like Elasticsearch when a requested snapshot index is still open on the target.
+func (m *mockESClientForRestore) RestoreSnapshot(_, snapshotName, targets string, partial bool) error {
 	if m.restoreErr != nil {
 		return m.restoreErr
 	}
+	for _, target := range strings.Split(targets, ",") {
+		if m.snapshot != nil && !slices.Contains(m.snapshot.Indices, target) && !slices.Contains(m.snapshot.DataStreams, target) {
+			return fmt.Errorf("index_not_found_exception: no such index [%s]", target)
+		}
+		if slices.Contains(m.indices, target) {
+			return fmt.Errorf("cannot restore index [%s] because an open index with same name already exists", target)
+		}
+	}
 	m.restoredSnapshot = snapshotName
+	m.restoredTargets = targets
+	m.restoredPartial = partial
+	m.restoredDatastreams = map[string][]string{}
+	m.restoredPlain = nil
+	for _, target := range strings.Split(targets, ",") {
+		if !slices.Contains(m.snapshot.DataStreams, target) {
+			m.restoredPlain = append(m.restoredPlain, target)
+		}
+	}
+	for _, name := range m.snapshot.DataStreams {
+		if slices.Contains(strings.Split(targets, ","), name) {
+			// Like Elasticsearch, naming the stream restores all of its snapshotted backing indices.
+			for _, index := range m.snapshot.Indices {
+				if strings.HasPrefix(index, ".ds-"+name+"-") && !slices.Contains(m.restoredPlain, index) {
+					m.restoredPlain = append(m.restoredPlain, index)
+				}
+			}
+			m.restoredDatastreams[name] = slices.DeleteFunc(slices.Clone(m.restoredPlain), func(i string) bool {
+				return !strings.HasPrefix(i, ".ds-"+name+"-")
+			})
+			m.restoredPlain = slices.DeleteFunc(m.restoredPlain, func(i string) bool {
+				return strings.HasPrefix(i, ".ds-"+name+"-")
+			})
+		}
+	}
 	return nil
 }
 
-func (m *mockESClientForRestore) RolloverDatastream(datastreamName string) error {
-	if m.rolloverErr != nil {
-		return m.rolloverErr
+func (m *mockESClientForRestore) DeleteDatastream(datastreamName string) (bool, error) {
+	m.datastreamDeleteCalls = append(m.datastreamDeleteCalls, datastreamName)
+	if m.deleteDatastreamErr != nil {
+		return false, m.deleteDatastreamErr
 	}
-	m.rolledOverDS = datastreamName
-	return nil
+	backing, ok := m.datastreams[datastreamName]
+	if !ok {
+		return false, nil
+	}
+	m.indices = slices.DeleteFunc(m.indices, func(i string) bool { return slices.Contains(backing, i) })
+	if m.indexExistsMap == nil {
+		m.indexExistsMap = make(map[string]bool)
+	}
+	for _, index := range backing {
+		m.indexExistsMap[index] = false
+	}
+	delete(m.datastreams, datastreamName)
+	return true, nil
 }
 
 func (m *mockESClientForRestore) ListSnapshots(_ string) ([]elasticsearch.Snapshot, error) {
@@ -295,6 +383,415 @@ func TestDeleteAllSTSIndicesHonorsIndicesPattern(t *testing.T) {
 	assert.Equal(t, []string{"sts_topology"}, client.deletedIndices)
 }
 
+func newLogsRestoreContext() *app.Context {
+	return &app.Context{
+		Config: &config.Config{
+			Elasticsearch: config.ElasticsearchConfig{
+				Restore: config.RestoreConfig{
+					IndexPrefix:           "sts",
+					DatastreamIndexPrefix: ".ds-sts_k8s_logs",
+					DatastreamName:        "sts_k8s_logs",
+					IndicesPattern:        "sts*,.ds-sts_k8s_logs*",
+				},
+			},
+		},
+		Logger: logger.New(true, false),
+	}
+}
+
+//nolint:funlen
+func TestReplaceSTSIndicesRestoresLogsDatastream(t *testing.T) {
+	const (
+		topo = "sts_topology_events-2026.11.01"
+		b1   = ".ds-sts_k8s_logs-2026.11.01-000001"
+		b2   = ".ds-sts_k8s_logs-2026.11.01-000002"
+		b3   = ".ds-sts_k8s_logs-2026.11.09-000003"
+		full = topo + "," + b1 + "," + b2 + ",sts_k8s_logs"
+	)
+	dsSnapshot := "sts_k8s_logs"
+
+	tests := []struct {
+		name                string
+		pattern             string
+		targetIndices       []string
+		targetDatastreams   map[string][]string
+		addOnListCall       map[int][]string
+		deleteDatastreamErr error
+		snapshotState       string
+		expectDSCalls       []string
+		expectDeleted       []string
+		expectTargets       string
+		expectRemaining     []string
+		expectListCalls     int
+		expectPartial       bool
+		expectErr           string
+	}{
+		{
+			name:            "T1 no STS indices",
+			targetIndices:   []string{},
+			expectTargets:   full,
+			expectRemaining: []string{},
+			expectListCalls: 2,
+		},
+		{
+			name:            "T2 plain indices only",
+			targetIndices:   []string{topo, "other"},
+			expectDeleted:   []string{topo},
+			expectTargets:   full,
+			expectRemaining: []string{"other"},
+		},
+		{
+			name:              "T3 live data stream write index collides",
+			targetIndices:     []string{topo, b2},
+			targetDatastreams: map[string][]string{dsSnapshot: {b2}},
+			expectDSCalls:     []string{dsSnapshot},
+			expectDeleted:     []string{topo},
+			expectTargets:     full,
+			expectRemaining:   []string{},
+			expectListCalls:   3,
+		},
+		{
+			name:          "T4 orphaned backing indices, no data stream",
+			targetIndices: []string{topo, b1},
+			expectDSCalls: []string{dsSnapshot},
+			expectDeleted: []string{topo, b1},
+			expectTargets: full,
+		},
+		{
+			name:              "T5 data stream plus orphans",
+			targetIndices:     []string{topo, b1, b3},
+			targetDatastreams: map[string][]string{dsSnapshot: {b3}},
+			expectDSCalls:     []string{dsSnapshot},
+			expectDeleted:     []string{topo, b1},
+			expectTargets:     full,
+			expectRemaining:   []string{},
+		},
+		{
+			name:              "T7 pattern excludes logs",
+			pattern:           "sts_topology*",
+			targetIndices:     []string{topo, b2},
+			targetDatastreams: map[string][]string{dsSnapshot: {b2}},
+			expectDeleted:     []string{topo},
+			expectTargets:     topo,
+			expectRemaining:   []string{b2},
+		},
+		{
+			name:          "T8 writer recreates the data stream",
+			targetIndices: []string{topo},
+			addOnListCall: map[int][]string{2: {".ds-sts_k8s_logs-2026.11.09-000001"}},
+			expectDeleted: []string{topo},
+			expectErr:     "reappeared after deletion",
+		},
+		{
+			name:                "T9 data stream delete fails",
+			targetIndices:       []string{b1},
+			deleteDatastreamErr: fmt.Errorf("boom"),
+			expectDSCalls:       []string{dsSnapshot},
+			expectErr:           "failed to delete datastream",
+			expectRemaining:     []string{b1},
+		},
+		{
+			name:              "T10 partial is passed through",
+			targetIndices:     []string{topo, b2},
+			targetDatastreams: map[string][]string{dsSnapshot: {b2}},
+			snapshotState:     "PARTIAL",
+			expectDSCalls:     []string{dsSnapshot},
+			expectDeleted:     []string{topo},
+			expectTargets:     full,
+			expectRemaining:   []string{},
+			expectPartial:     true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			state := tt.snapshotState
+			if state == "" {
+				state = elasticsearch.StatusSuccess
+			}
+			snapshot := &elasticsearch.Snapshot{
+				Snapshot:    "sts-backup-20261101",
+				State:       state,
+				Indices:     []string{topo, b1, b2},
+				DataStreams: []string{dsSnapshot},
+			}
+			client := &mockESClientForRestore{
+				indices:             slices.Clone(tt.targetIndices),
+				snapshot:            snapshot,
+				datastreams:         tt.targetDatastreams,
+				addOnListCall:       tt.addOnListCall,
+				deleteDatastreamErr: tt.deleteDatastreamErr,
+				indexExistsMap:      map[string]bool{},
+			}
+			for _, index := range tt.targetIndices {
+				client.indexExistsMap[index] = true
+			}
+			appCtx := newLogsRestoreContext()
+			if tt.pattern != "" {
+				appCtx.Config.Elasticsearch.Restore.IndicesPattern = tt.pattern
+			}
+			restoreIndices, err := restorableSnapshotIndices(snapshot, appCtx.Config.Elasticsearch.Restore)
+			require.NoError(t, err)
+			targets, err := restoreTargets(snapshot, restoreIndices, appCtx.Config.Elasticsearch.Restore)
+			require.NoError(t, err)
+
+			err = replaceSTSIndices(client, appCtx, "backup-repo", snapshot, targets)
+
+			assert.Equal(t, tt.expectDSCalls, client.datastreamDeleteCalls)
+			assert.Equal(t, tt.expectDeleted, client.deletedIndices)
+			if tt.expectErr != "" {
+				require.ErrorContains(t, err, tt.expectErr)
+				assert.Empty(t, client.restoredTargets, "restore must not be sent")
+				if tt.name == "T8 writer recreates the data stream" {
+					assert.ErrorContains(t, err, ".ds-sts_k8s_logs-2026.11.09-000001")
+				}
+				if tt.expectRemaining != nil {
+					assert.Equal(t, tt.expectRemaining, client.indices)
+				}
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectTargets, client.restoredTargets)
+			assert.Equal(t, tt.expectPartial, client.restoredPartial)
+			if tt.expectRemaining != nil {
+				assert.Equal(t, tt.expectRemaining, client.indices)
+			}
+			if tt.expectListCalls != 0 {
+				assert.Equal(t, tt.expectListCalls, client.listCalls)
+			}
+		})
+	}
+}
+
+func TestRestoreTargetsRejectsSnapshotWithoutDatastream(t *testing.T) {
+	restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore
+	const topo = "sts_topology_events-2026.11.01"
+
+	_, err := restoreTargets(&elasticsearch.Snapshot{Snapshot: "snap"},
+		[]string{topo, ".ds-sts_k8s_logs-2026.11.01-000001"}, restoreConfig)
+	require.ErrorContains(t, err, "not the sts_k8s_logs data stream")
+
+	targets, err := restoreTargets(&elasticsearch.Snapshot{Snapshot: "snap"}, []string{topo}, restoreConfig)
+	require.NoError(t, err)
+	assert.Equal(t, []string{topo}, targets)
+}
+
+func TestPrepareRestoreTargetsRequiresAllOrNoLogsBackingIndices(t *testing.T) {
+	const (
+		topo = "sts_topology_events-2026.11.01"
+		b1   = ".ds-sts_k8s_logs-2026.11.01-000001"
+		b2   = ".ds-sts_k8s_logs-2026.11.01-000002"
+	)
+	snapshot := &elasticsearch.Snapshot{
+		Snapshot: "snap", State: elasticsearch.StatusSuccess,
+		Indices: []string{topo, b1, b2}, DataStreams: []string{"sts_k8s_logs"},
+	}
+
+	tests := []struct {
+		name          string
+		pattern       string
+		expectErr     string
+		expectTargets []string
+	}{
+		{name: "one of two backing indices", pattern: "sts*,.ds-sts_k8s_logs-2026.11.01-000002", expectErr: "omitted: " + b1},
+		{name: "exclusion of a backing index", pattern: "sts*,.ds-sts_k8s_logs*,-" + b1, expectErr: "omitted: " + b1},
+		{name: "all backing indices", pattern: "sts*,.ds-sts_k8s_logs*", expectTargets: []string{topo, b1, b2, "sts_k8s_logs"}},
+		{name: "no backing indices", pattern: "sts*", expectTargets: []string{topo}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore
+			restoreConfig.IndicesPattern = tt.pattern
+
+			targets, err := prepareRestoreTargets(snapshot, restoreConfig)
+
+			if tt.expectErr != "" {
+				require.ErrorContains(t, err, tt.expectErr)
+				assert.ErrorContains(t, err, "must select all of them or none")
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectTargets, targets)
+		})
+	}
+}
+
+func TestMockRestoreExpandsNamedDatastream(t *testing.T) {
+	const (
+		b1 = ".ds-sts_k8s_logs-2026.11.01-000001"
+		b2 = ".ds-sts_k8s_logs-2026.11.01-000002"
+	)
+	client := &mockESClientForRestore{
+		snapshot:       &elasticsearch.Snapshot{Indices: []string{b1, b2}, DataStreams: []string{"sts_k8s_logs"}},
+		indexExistsMap: map[string]bool{},
+	}
+
+	require.NoError(t, client.RestoreSnapshot("repo", "snap", b2+",sts_k8s_logs", false))
+
+	assert.Equal(t, []string{b2, b1}, client.restoredDatastreams["sts_k8s_logs"])
+	assert.Empty(t, client.restoredPlain)
+}
+
+func TestPrepareRestoreTargetsRejectsBeforeAnyClusterAccess(t *testing.T) {
+	restoreConfig := newLogsRestoreContext().Config.Elasticsearch.Restore
+	snapshot := &elasticsearch.Snapshot{
+		Snapshot: "snap",
+		State:    elasticsearch.StatusSuccess,
+		Indices:  []string{"sts_topology_events-2026.11.01", ".ds-sts_k8s_logs-2026.11.01-000001"},
+	}
+
+	targets, err := prepareRestoreTargets(snapshot, restoreConfig)
+
+	require.ErrorContains(t, err, "not the sts_k8s_logs data stream")
+	assert.Nil(t, targets)
+}
+
+func TestVerifyLogsDatastream(t *testing.T) {
+	const (
+		topo = "sts_topology_events-2026.11.01"
+		b1   = ".ds-sts_k8s_logs-2026.11.01-000001"
+		b2   = ".ds-sts_k8s_logs-2026.11.01-000002"
+	)
+	appCtx := newLogsRestoreContext()
+	restoreConfig := appCtx.Config.Elasticsearch.Restore
+	snapshot := &elasticsearch.Snapshot{
+		Snapshot: "snap", State: elasticsearch.StatusSuccess,
+		Indices: []string{topo, b1, b2}, DataStreams: []string{"sts_k8s_logs"},
+	}
+	expected, err := restorableSnapshotIndices(snapshot, restoreConfig)
+	require.NoError(t, err)
+
+	t.Run("restored as a data stream", func(t *testing.T) {
+		client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}}
+		targets, err := restoreTargets(snapshot, expected, restoreConfig)
+		require.NoError(t, err)
+		require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(targets, ","), false))
+		client.settle()
+
+		assert.NoError(t, verifyLogsDatastream(client, appCtx.Logger, restoreConfig, expected))
+	})
+
+	t.Run("restored as plain indices", func(t *testing.T) {
+		client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}}
+		require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(expected, ","), false))
+		client.settle()
+
+		err := verifyLogsDatastream(client, appCtx.Logger, restoreConfig, expected)
+
+		require.ErrorContains(t, err, "'sts_k8s_logs' data stream does not exist")
+	})
+
+	t.Run("backing index outside the data stream", func(t *testing.T) {
+		client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}}
+		targets, err := restoreTargets(snapshot, expected, restoreConfig)
+		require.NoError(t, err)
+		require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(targets, ","), false))
+		client.settle()
+		client.indices = append(client.indices, ".ds-sts_k8s_logs-2026.11.09-000001")
+
+		err = verifyLogsDatastream(client, appCtx.Logger, restoreConfig, expected)
+
+		require.ErrorContains(t, err, "outside the data stream: .ds-sts_k8s_logs-2026.11.09-000001")
+	})
+
+	t.Run("restore without logs indices is not checked", func(t *testing.T) {
+		client := &mockESClientForRestore{snapshot: snapshot}
+
+		assert.NoError(t, verifyLogsDatastream(client, appCtx.Logger, restoreConfig, []string{topo}))
+	})
+}
+
+type flakyDatastreamReader struct {
+	failures int
+	calls    int
+	backing  []string
+}
+
+func (f *flakyDatastreamReader) GetDatastream(string) ([]string, bool, error) {
+	f.calls++
+	if f.calls <= f.failures {
+		return nil, false, fmt.Errorf("connection refused")
+	}
+	return f.backing, true, nil
+}
+
+func (f *flakyDatastreamReader) ListIndices(string) ([]string, error) {
+	return f.backing, nil
+}
+
+func TestVerifyLogsDatastreamRetriesTransientErrors(t *testing.T) {
+	verifyRetryInterval = 0
+	t.Cleanup(func() { verifyRetryInterval = 2 * time.Second })
+	appCtx := newLogsRestoreContext()
+	restoreConfig := appCtx.Config.Elasticsearch.Restore
+	expected := []string{"sts_topology_events-2026.11.01", ".ds-sts_k8s_logs-2026.11.01-000001"}
+	backing := []string{".ds-sts_k8s_logs-2026.11.01-000001"}
+
+	t.Run("recovers after a dropped connection", func(t *testing.T) {
+		reader := &flakyDatastreamReader{failures: verifyAttempts - 1, backing: backing}
+
+		assert.NoError(t, verifyLogsDatastream(reader, appCtx.Logger, restoreConfig, expected))
+		assert.Equal(t, verifyAttempts, reader.calls)
+	})
+
+	t.Run("gives up after the last attempt", func(t *testing.T) {
+		reader := &flakyDatastreamReader{failures: verifyAttempts, backing: backing}
+
+		err := verifyLogsDatastream(reader, appCtx.Logger, restoreConfig, expected)
+
+		require.ErrorContains(t, err, "failed to verify datastream")
+		assert.Equal(t, verifyAttempts, reader.calls)
+	})
+}
+
+func TestVerifyAndFinalizeFailsWithFinalizeHint(t *testing.T) {
+	appCtx := newLogsRestoreContext()
+	appCtx.Namespace = "ns"
+	snapshot := &elasticsearch.Snapshot{
+		Snapshot: "snap", State: elasticsearch.StatusSuccess,
+		Indices:     []string{"sts_topology_events-2026.11.01", ".ds-sts_k8s_logs-2026.11.01-000001"},
+		DataStreams: []string{"sts_k8s_logs"},
+	}
+	client := &mockESClientForRestore{snapshot: snapshot, indexExistsMap: map[string]bool{}}
+	require.NoError(t, client.RestoreSnapshot("repo", "snap", strings.Join(snapshot.Indices, ","), false))
+	client.settle()
+
+	err := verifyAndFinalize(client, appCtx, snapshot.Indices)
+
+	require.ErrorContains(t, err, "does not exist")
+	assert.ErrorContains(t, err, "check-and-finalize --finalize-only -n ns")
+}
+
+func TestRestoreTargetsKeepCompletionExpectation(t *testing.T) {
+	const (
+		topo = "sts_topology_events-2026.11.01"
+		b1   = ".ds-sts_k8s_logs-2026.11.01-000001"
+		b2   = ".ds-sts_k8s_logs-2026.11.01-000002"
+	)
+	appCtx := newLogsRestoreContext()
+	restoreConfig := appCtx.Config.Elasticsearch.Restore
+	snapshot := &elasticsearch.Snapshot{
+		Snapshot:    "snap",
+		Indices:     []string{topo, b1, b2},
+		DataStreams: []string{"sts_k8s_logs"},
+	}
+
+	expected, err := restorableSnapshotIndices(snapshot, restoreConfig)
+	require.NoError(t, err)
+	targets, err := restoreTargets(snapshot, expected, restoreConfig)
+	require.NoError(t, err)
+	fromCheck, err := expectedRestoredIndices(&snapshotOnlyClient{snapshot: snapshot}, appCtx, "repo", "snap")
+	require.NoError(t, err)
+
+	assert.Equal(t, expected, fromCheck)
+	assert.NotContains(t, expected, "sts_k8s_logs")
+	assert.Equal(t, append(slices.Clone(expected), "sts_k8s_logs"), targets)
+
+	restored := elasticsearch.IndexHealth{NumberOfShards: 1, ActivePrimaryShards: 1}
+	health := map[string]elasticsearch.IndexHealth{topo: restored, b1: restored, b2: restored}
+	assert.True(t, measureRestore(expected, restoreConfig.DatastreamIndexPrefix, health).complete())
+}
+
 // TestHasDatastreamIndices tests datastream detection
 func TestHasDatastreamIndices(t *testing.T) {
 	tests := []struct {
@@ -357,10 +854,8 @@ func TestMockESClientForRestore(t *testing.T) {
 		indicesToDelete    []string
 		deleteErr          error
 		restoreErr         error
-		rolloverErr        error
 		expectDeletedCount int
 		expectRestoreOK    bool
-		expectRolloverOK   bool
 	}{
 		{
 			name:               "successful restore with index deletion",
@@ -398,14 +893,6 @@ func TestMockESClientForRestore(t *testing.T) {
 			expectDeletedCount: 0,
 			expectRestoreOK:    false,
 		},
-		{
-			name:               "successful rollover",
-			initialIndices:     []string{"sts_k8s_logs-000001"},
-			indicesToDelete:    []string{},
-			rolloverErr:        nil,
-			expectDeletedCount: 0,
-			expectRolloverOK:   true,
-		},
 	}
 
 	for _, tt := range tests {
@@ -414,7 +901,6 @@ func TestMockESClientForRestore(t *testing.T) {
 				indices:        tt.initialIndices,
 				deleteErr:      tt.deleteErr,
 				restoreErr:     tt.restoreErr,
-				rolloverErr:    tt.rolloverErr,
 				indexExistsMap: make(map[string]bool),
 				snapshot: &elasticsearch.Snapshot{
 					Snapshot:   "test-snapshot",
@@ -441,15 +927,8 @@ func TestMockESClientForRestore(t *testing.T) {
 
 			assert.Equal(t, tt.expectDeletedCount, len(mockClient.deletedIndices))
 
-			// Test rollover if applicable
-			if tt.expectRolloverOK {
-				err := mockClient.RolloverDatastream("sts_k8s_logs")
-				assert.NoError(t, err)
-				assert.Equal(t, "sts_k8s_logs", mockClient.rolledOverDS)
-			}
-
 			// Test restore
-			err := mockClient.RestoreSnapshot("backup-repo", "test-snapshot", "sts_*", false)
+			err := mockClient.RestoreSnapshot("backup-repo", "test-snapshot", "sts_topology,sts_metrics", false)
 			if tt.expectRestoreOK {
 				assert.NoError(t, err)
 				assert.Equal(t, "test-snapshot", mockClient.restoredSnapshot)
diff --git a/internal/clients/elasticsearch/client.go b/internal/clients/elasticsearch/client.go
index f18aed7..b689e54 100644
--- a/internal/clients/elasticsearch/client.go
+++ b/internal/clients/elasticsearch/client.go
@@ -60,6 +60,7 @@ type Snapshot struct {
 	EndTimeMillis    int64             `json:"end_time_in_millis"`
 	DurationInMillis int64             `json:"duration_in_millis"`
 	Indices          []string          `json:"indices"`
+	DataStreams      []string          `json:"data_streams"`
 	Failures         []SnapshotFailure `json:"failures"`
 	Shards           struct {
 		Total      int `json:"total"`
@@ -240,22 +241,73 @@ func (c *Client) IndexExists(index string) (bool, error) {
 	return true, nil
 }
 
-// RolloverDatastream performs a rollover on a datastream
-func (c *Client) RolloverDatastream(datastreamName string) error {
-	res, err := c.es.Indices.Rollover(
-		datastreamName,
-		c.es.Indices.Rollover.WithContext(context.Background()),
+// DeleteDatastream deletes a data stream together with all of its backing indices.
+// It reports false when the data stream does not exist.
+func (c *Client) DeleteDatastream(datastreamName string) (bool, error) {
+	res, err := c.es.Indices.DeleteDataStream(
+		[]string{datastreamName},
+		c.es.Indices.DeleteDataStream.WithContext(context.Background()),
 	)
 	if err != nil {
-		return fmt.Errorf("failed to rollover datastream: %w", err)
+		return false, fmt.Errorf("failed to delete datastream: %w", err)
 	}
 	defer res.Body.Close()
 
+	if res.StatusCode == http.StatusNotFound {
+		return false, nil
+	}
+
 	if res.IsError() {
-		return fmt.Errorf("elasticsearch returned error: %s", res.String())
+		return false, fmt.Errorf("elasticsearch returned error: %s", res.String())
 	}
 
-	return nil
+	return true, nil
+}
+
+// GetDatastream returns the names of a data stream's backing indices.
+// It reports false when the data stream does not exist.
+func (c *Client) GetDatastream(datastreamName string) ([]string, bool, error) {
+	res, err := c.es.Indices.GetDataStream(
+		c.es.Indices.GetDataStream.WithName(datastreamName),
+		c.es.Indices.GetDataStream.WithContext(context.Background()),
+	)
+	if err != nil {
+		return nil, false, fmt.Errorf("failed to get datastream: %w", err)
+	}
+	defer res.Body.Close()
+
+	if res.StatusCode == http.StatusNotFound {
+		return nil, false, nil
+	}
+
+	if res.IsError() {
+		return nil, false, fmt.Errorf("elasticsearch returned error: %s", res.String())
+	}
+
+	var body struct {
+		DataStreams []struct {
+			Name    string `json:"name"`
+			Indices []struct {
+				IndexName string `json:"index_name"`
+			} `json:"indices"`
+		} `json:"data_streams"`
+	}
+	if err := json.NewDecoder(res.Body).Decode(&body); err != nil {
+		return nil, false, fmt.Errorf("failed to decode response: %w", err)
+	}
+
+	for _, ds := range body.DataStreams {
+		if ds.Name != datastreamName {
+			continue
+		}
+		indices := make([]string, len(ds.Indices))
+		for i, idx := range ds.Indices {
+			indices[i] = idx.IndexName
+		}
+		return indices, true, nil
+	}
+
+	return nil, false, nil
 }
 
 // DeleteSnapshotRepository deletes a snapshot repository
diff --git a/internal/clients/elasticsearch/client_test.go b/internal/clients/elasticsearch/client_test.go
index 5392aab..82b7d6a 100644
--- a/internal/clients/elasticsearch/client_test.go
+++ b/internal/clients/elasticsearch/client_test.go
@@ -134,7 +134,8 @@ func TestClient_GetSnapshot(t *testing.T) {
 						"uuid": "uuid-1",
 						"repository": "test-repo",
 						"state": "SUCCESS",
-						"indices": ["index-1", "index-2"]
+						"indices": ["index-1", "index-2"],
+						"data_streams": ["sts_k8s_logs"]
 					}
 				]
 			}`,
@@ -181,6 +182,7 @@ func TestClient_GetSnapshot(t *testing.T) {
 			assert.NotNil(t, snapshot)
 			assert.Equal(t, tt.snapshotName, snapshot.Snapshot)
 			assert.Equal(t, tt.repository, snapshot.Repository)
+			assert.Equal(t, []string{"sts_k8s_logs"}, snapshot.DataStreams)
 		})
 	}
 }
@@ -309,6 +311,92 @@ func TestClient_DeleteIndex(t *testing.T) {
 	}
 }
 
+func TestClient_DeleteDatastream(t *testing.T) {
+	tests := []struct {
+		name           string
+		responseStatus int
+		expectDeleted  bool
+		expectError    bool
+	}{
+		{name: "deleted", responseStatus: http.StatusOK, expectDeleted: true},
+		{name: "not found", responseStatus: http.StatusNotFound, expectDeleted: false},
+		{name: "elasticsearch error", responseStatus: http.StatusInternalServerError, expectError: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := mockESServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				assert.Equal(t, "/_data_stream/sts_k8s_logs", r.URL.Path)
+				assert.Equal(t, http.MethodDelete, r.Method)
+
+				w.WriteHeader(tt.responseStatus)
+			}))
+			defer server.Close()
+
+			client, err := NewClient(server.URL)
+			require.NoError(t, err)
+
+			deleted, err := client.DeleteDatastream("sts_k8s_logs")
+
+			if tt.expectError {
+				assert.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectDeleted, deleted)
+		})
+	}
+}
+
+func TestClient_GetDatastream(t *testing.T) {
+	tests := []struct {
+		name           string
+		responseStatus int
+		responseBody   string
+		expectIndices  []string
+		expectExists   bool
+		expectError    bool
+	}{
+		{
+			name:           "exists",
+			responseStatus: http.StatusOK,
+			responseBody:   `{"data_streams":[{"name":"sts_k8s_logs","indices":[{"index_name":".ds-sts_k8s_logs-1"},{"index_name":".ds-sts_k8s_logs-2"}]}]}`,
+			expectIndices:  []string{".ds-sts_k8s_logs-1", ".ds-sts_k8s_logs-2"},
+			expectExists:   true,
+		},
+		{name: "not found", responseStatus: http.StatusNotFound, responseBody: `{}`},
+		{name: "empty result", responseStatus: http.StatusOK, responseBody: `{"data_streams":[]}`},
+		{name: "elasticsearch error", responseStatus: http.StatusInternalServerError, responseBody: `{}`, expectError: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := mockESServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				assert.Equal(t, "/_data_stream/sts_k8s_logs", r.URL.Path)
+				assert.Equal(t, http.MethodGet, r.Method)
+
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(tt.responseStatus)
+				_, _ = w.Write([]byte(tt.responseBody))
+			}))
+			defer server.Close()
+
+			client, err := NewClient(server.URL)
+			require.NoError(t, err)
+
+			indices, exists, err := client.GetDatastream("sts_k8s_logs")
+
+			if tt.expectError {
+				assert.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectExists, exists)
+			assert.Equal(t, tt.expectIndices, indices)
+		})
+	}
+}
+
 func TestClient_IndexExists(t *testing.T) {
 	tests := []struct {
 		name           string
diff --git a/internal/clients/elasticsearch/interface.go b/internal/clients/elasticsearch/interface.go
index f8e0089..cd66417 100644
--- a/internal/clients/elasticsearch/interface.go
+++ b/internal/clients/elasticsearch/interface.go
@@ -16,7 +16,8 @@ type Interface interface {
 	IndexExists(index string) (bool, error)
 
 	// Datastream operations
-	RolloverDatastream(datastreamName string) error
+	DeleteDatastream(datastreamName string) (bool, error)
+	GetDatastream(datastreamName string) ([]string, bool, error)
 
 	// Repository and SLM operations
 	DeleteSnapshotRepository(name string) error
diff --git a/internal/clients/k8s/client.go b/internal/clients/k8s/client.go
index e955666..fd05b03 100644
--- a/internal/clients/k8s/client.go
+++ b/internal/clients/k8s/client.go
@@ -10,6 +10,7 @@ import (
 	"net/url"
 	"os"
 	"path"
+	"strconv"
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -202,6 +203,8 @@ const (
 type AppsScale struct {
 	Name     string
 	Replicas int32
+	// PodSelector selects the workload's pods (from spec.selector); only set by scale-down.
+	PodSelector string
 }
 
 // RestoreLockInfo holds information about an active restore lock on a resource
@@ -274,6 +277,10 @@ func scaleDownDeployment(ctx context.Context, client kubernetes.Interface, names
 		if dep.Annotations == nil {
 			dep.Annotations = make(map[string]string)
 		}
+		if recorded, ok := recordedReplicas(dep.Annotations, originalReplicas); ok {
+			originalReplicas = recorded
+			return nil
+		}
 		dep.Annotations[PreRestoreReplicasAnnotation] = fmt.Sprintf("%d", originalReplicas)
 		zero := int32(0)
 		dep.Spec.Replicas = &zero
@@ -297,6 +304,10 @@ func scaleDownStatefulSet(ctx context.Context, client kubernetes.Interface, name
 		if sts.Annotations == nil {
 			sts.Annotations = make(map[string]string)
 		}
+		if recorded, ok := recordedReplicas(sts.Annotations, originalReplicas); ok {
+			originalReplicas = recorded
+			return nil
+		}
 		sts.Annotations[PreRestoreReplicasAnnotation] = fmt.Sprintf("%d", originalReplicas)
 		zero := int32(0)
 		sts.Spec.Replicas = &zero
@@ -376,6 +387,30 @@ func scaleUpStatefulSetFromAnnotation(ctx context.Context, client kubernetes.Int
 	return scaledTo, found, err
 }
 
+// podSelectorString converts a workload's spec.selector into a pod label selector string.
+// The restore label sits on the workload only, so pods must be found through spec.selector.
+func podSelectorString(name string, selector *metav1.LabelSelector) (string, error) {
+	sel, err := metav1.LabelSelectorAsSelector(selector)
+	if err != nil {
+		return "", fmt.Errorf("invalid selector on %s: %w", name, err)
+	}
+	return sel.String(), nil
+}
+
+// recordedReplicas returns the replica count saved by an earlier scale-down when the workload is
+// already at 0. A retry after a failed scale-down must not overwrite it with 0, or scale-up
+// would restore 0 replicas.
+func recordedReplicas(annotations map[string]string, currentReplicas int32) (int32, bool) {
+	if currentReplicas != 0 {
+		return 0, false
+	}
+	recorded, err := strconv.ParseInt(annotations[PreRestoreReplicasAnnotation], 10, 32)
+	if err != nil {
+		return 0, false
+	}
+	return int32(recorded), true
+}
+
 // ScaleDownDeployments scales down deployments matching a label selector to 0 replicas
 // Returns a list of deployment names and their original replica counts
 func (c *Client) ScaleDownDeployments(namespace, labelSelector string) ([]AppsScale, error) {
@@ -391,13 +426,18 @@ func (c *Client) ScaleDownDeployments(namespace, labelSelector string) ([]AppsSc
 
 	var scaledResources []AppsScale
 	for _, dep := range deployments.Items {
+		podSelector, err := podSelectorString(dep.Name, dep.Spec.Selector)
+		if err != nil {
+			return scaledResources, err
+		}
 		originalReplicas, err := scaleDownDeployment(ctx, c.clientset, namespace, dep.Name)
 		if err != nil {
 			return scaledResources, fmt.Errorf("failed to scale down deployment %s: %w", dep.Name, err)
 		}
 		scaledResources = append(scaledResources, AppsScale{
-			Name:     dep.Name,
-			Replicas: originalReplicas,
+			Name:        dep.Name,
+			Replicas:    originalReplicas,
+			PodSelector: podSelector,
 		})
 	}
 
@@ -454,13 +494,18 @@ func (c *Client) ScaleDownStatefulSets(namespace, labelSelector string) ([]AppsS
 
 	var scaledResources []AppsScale
 	for _, sts := range statefulSets.Items {
+		podSelector, err := podSelectorString(sts.Name, sts.Spec.Selector)
+		if err != nil {
+			return scaledResources, err
+		}
 		originalReplicas, err := scaleDownStatefulSet(ctx, c.clientset, namespace, sts.Name)
 		if err != nil {
 			return scaledResources, fmt.Errorf("failed to scale down statefulset %s: %w", sts.Name, err)
 		}
 		scaledResources = append(scaledResources, AppsScale{
-			Name:     sts.Name,
-			Replicas: originalReplicas,
+			Name:        sts.Name,
+			Replicas:    originalReplicas,
+			PodSelector: podSelector,
 		})
 	}
 
diff --git a/internal/clients/k8s/client_test.go b/internal/clients/k8s/client_test.go
index cefd882..9d9e105 100644
--- a/internal/clients/k8s/client_test.go
+++ b/internal/clients/k8s/client_test.go
@@ -965,3 +965,30 @@ func TestClient_SetAndClearRestoreLock_Integration(t *testing.T) {
 	require.NoError(t, err)
 	assert.Empty(t, locks)
 }
+
+func TestClient_ScaleDownReturnsPodSelector(t *testing.T) {
+	fakeClient := fake.NewSimpleClientset()
+	client := &Client{clientset: fakeClient}
+	ctx := context.Background()
+	meta := map[string]string{"restore": "true"}
+	selector := &metav1.LabelSelector{MatchLabels: map[string]string{"app": "x", "tier": "y"}}
+
+	deploy := createDeployment("deploy", "test-ns", meta, 1)
+	deploy.Spec.Selector = selector
+	_, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{})
+	require.NoError(t, err)
+	sts := createStatefulSet("sts", "test-ns", meta, 1)
+	sts.Spec.Selector = selector
+	_, err = fakeClient.AppsV1().StatefulSets("test-ns").Create(ctx, &sts, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	deployScales, err := client.ScaleDownDeployments("test-ns", "restore=true")
+	require.NoError(t, err)
+	require.Len(t, deployScales, 1)
+	assert.Equal(t, "app=x,tier=y", deployScales[0].PodSelector)
+
+	stsScales, err := client.ScaleDownStatefulSets("test-ns", "restore=true")
+	require.NoError(t, err)
+	require.Len(t, stsScales, 1)
+	assert.Equal(t, "app=x,tier=y", stsScales[0].PodSelector)
+}
diff --git a/internal/foundation/config/testdata/validConfigMapConfig.yaml b/internal/foundation/config/testdata/validConfigMapConfig.yaml
index 4308734..c58b2ee 100644
--- a/internal/foundation/config/testdata/validConfigMapConfig.yaml
+++ b/internal/foundation/config/testdata/validConfigMapConfig.yaml
@@ -54,7 +54,7 @@ elasticsearch:
     indexPrefix: sts
     # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation})
     datastreamIndexPrefix: .ds-sts_k8s_logs
-    # Name of the datastream (used for rollover operations)
+    # Name of the datastream (deleted and restored together with its backing indices)
     datastreamName: sts_k8s_logs
     # Pattern for indices to restore from snapshot (comma-separated glob patterns)
     indicesPattern: sts*,.ds-sts_k8s_logs*
diff --git a/internal/foundation/config/testdata/validConfigMapOnly.yaml b/internal/foundation/config/testdata/validConfigMapOnly.yaml
index 3003119..072090a 100644
--- a/internal/foundation/config/testdata/validConfigMapOnly.yaml
+++ b/internal/foundation/config/testdata/validConfigMapOnly.yaml
@@ -61,7 +61,7 @@ elasticsearch:
     indexPrefix: sts
     # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation})
     datastreamIndexPrefix: .ds-sts_k8s_logs
-    # Name of the datastream (used for rollover operations)
+    # Name of the datastream (deleted and restored together with its backing indices)
     datastreamName: sts_k8s_logs
     # Pattern for indices to restore from snapshot (comma-separated glob patterns)
     indicesPattern: sts*,.ds-sts_k8s_logs*
diff --git a/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml b/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml
index dc2f72e..5db8859 100644
--- a/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml
+++ b/internal/foundation/config/testdata/validStorageConfigMapConfig.yaml
@@ -56,7 +56,7 @@ elasticsearch:
     indexPrefix: sts
     # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation})
     datastreamIndexPrefix: .ds-sts_k8s_logs
-    # Name of the datastream (used for rollover operations)
+    # Name of the datastream (deleted and restored together with its backing indices)
     datastreamName: sts_k8s_logs
     # Pattern for indices to restore from snapshot (comma-separated glob patterns)
     indicesPattern: sts*,.ds-sts_k8s_logs*
diff --git a/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml b/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml
index ae82bcf..b8f3c12 100644
--- a/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml
+++ b/internal/foundation/config/testdata/validStorageConfigMapOnly.yaml
@@ -63,7 +63,7 @@ elasticsearch:
     indexPrefix: sts
     # Prefix for datastream indices (datastreams use pattern: .ds-{name}-{generation})
     datastreamIndexPrefix: .ds-sts_k8s_logs
-    # Name of the datastream (used for rollover operations)
+    # Name of the datastream (deleted and restored together with its backing indices)
     datastreamName: sts_k8s_logs
     # Pattern for indices to restore from snapshot (comma-separated glob patterns)
     indicesPattern: sts*,.ds-sts_k8s_logs*
diff --git a/internal/orchestration/scale/scale.go b/internal/orchestration/scale/scale.go
index 107a9ae..712177f 100644
--- a/internal/orchestration/scale/scale.go
+++ b/internal/orchestration/scale/scale.go
@@ -3,6 +3,7 @@ package scale
 import (
 	"context"
 	"fmt"
+	"slices"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -19,7 +20,7 @@ const (
 )
 
 // ScaleDown scales down deployments matching the label selector and logs the results.
-// It waits for all pods to terminate before returning.
+// It waits until no pod selected by any scaled workload remains.
 //
 //nolint:revive // Package name "scale" with function "ScaleDown" is intentionally verbose for clarity
 func ScaleDown(k8sClient *k8s.Client, namespace, labelSelector string, log *logger.Logger) ([]k8s.AppsScale, error) {
@@ -54,8 +55,16 @@ func ScaleDown(k8sClient *k8s.Client, namespace, labelSelector string, log *logg
 		log.Infof("  - %s (replicas: %d -> 0)", dep.Name, dep.Replicas)
 	}
 
-	// Wait for pods to terminate
-	if err := waitForPodsToTerminate(k8sClient, namespace, labelSelector, log); err != nil {
+	var podSelectors []string
+	for _, app := range scaledApps {
+		if app.PodSelector != "" {
+			podSelectors = append(podSelectors, app.PodSelector)
+		}
+	}
+	slices.Sort(podSelectors)
+	podSelectors = slices.Compact(podSelectors)
+
+	if err := waitForPodsToTerminate(k8sClient, namespace, podSelectors, log); err != nil {
 		return scaledApps, fmt.Errorf("failed waiting for pods to terminate: %w", err)
 	}
 
@@ -141,8 +150,8 @@ func ScaleUpAndReleaseLock(k8sClient *k8s.Client, namespace, labelSelector strin
 	return nil
 }
 
-// waitForPodsToTerminate polls for pod termination until all pods matching the label selector are gone
-func waitForPodsToTerminate(k8sClient *k8s.Client, namespace, labelSelector string, log *logger.Logger) error {
+// waitForPodsToTerminate polls until no pod matches any of the pod selectors
+func waitForPodsToTerminate(k8sClient *k8s.Client, namespace string, podSelectors []string, log *logger.Logger) error {
 	ctx := context.Background()
 	ticker := time.NewTicker(podTerminationCheckInterval)
 	defer ticker.Stop()
@@ -156,29 +165,28 @@ func waitForPodsToTerminate(k8sClient *k8s.Client, namespace, labelSelector stri
 		case <-timeout:
 			return fmt.Errorf("timeout waiting for pods to terminate after %v", podTerminationTimeout)
 		case <-ticker.C:
-			podList, err := k8sClient.Clientset().CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
-				LabelSelector: labelSelector,
-			})
-			if err != nil {
-				return fmt.Errorf("failed to list pods: %w", err)
-			}
-
-			// Filter out pods in terminal states (Succeeded, Failed) or being deleted
-			activePods := 0
-			for _, pod := range podList.Items {
-				if pod.Status.Phase != corev1.PodSucceeded &&
-					pod.Status.Phase != corev1.PodFailed &&
-					pod.DeletionTimestamp == nil {
-					activePods++
+			total := 0
+			for _, selector := range podSelectors {
+				podList, err := k8sClient.Clientset().CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
+					LabelSelector: selector,
+				})
+				if err != nil {
+					return fmt.Errorf("failed to list pods: %w", err)
+				}
+				for _, pod := range podList.Items {
+					// Finished pods (e.g. evicted) linger after scale-to-0 and never go away on their own.
+					if pod.Status.Phase != corev1.PodSucceeded && pod.Status.Phase != corev1.PodFailed {
+						total++
+					}
 				}
 			}
 
-			if activePods == 0 && len(podList.Items) == 0 {
+			if total == 0 {
 				log.Successf("All pods have terminated")
 				return nil
 			}
 
-			log.Infof("Waiting for %d pod(s) to terminate...", activePods+len(podList.Items)-activePods)
+			log.Infof("Waiting for %d pod(s) to terminate...", total)
 		}
 	}
 }
diff --git a/internal/orchestration/scale/scale_test.go b/internal/orchestration/scale/scale_test.go
index 2ee83f8..3725da2 100644
--- a/internal/orchestration/scale/scale_test.go
+++ b/internal/orchestration/scale/scale_test.go
@@ -13,6 +13,7 @@ import (
 	"github.com/stretchr/testify/require"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes/fake"
 )
@@ -122,7 +123,7 @@ func TestScaleDown_PodsInTerminalState(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	// Create pods in terminal states (should be ignored by waitForPodsToTerminate)
+	// Create pods in terminal states (ignored by waitForPodsToTerminate)
 	pod1 := createPod("pod1", map[string]string{"app": "test"}, corev1.PodSucceeded)
 	pod2 := createPod("pod2", map[string]string{"app": "test"}, corev1.PodFailed)
 
@@ -136,21 +137,10 @@ func TestScaleDown_PodsInTerminalState(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	// Delete terminal pods (simulating cleanup)
-	go func() {
-		time.Sleep(100 * time.Millisecond)
-		_ = fakeClient.CoreV1().Pods("test-ns").Delete(
-			context.Background(), "pod1", metav1.DeleteOptions{},
-		)
-		_ = fakeClient.CoreV1().Pods("test-ns").Delete(
-			context.Background(), "pod2", metav1.DeleteOptions{},
-		)
-	}()
-
 	// Execute scale down
 	scaledDeployments, err := ScaleDown(client, "test-ns", "app=test", log)
 
-	// Assertions - should succeed quickly since pods are in terminal states
+	// Finished pods are never deleted here: the wait must not count them
 	require.NoError(t, err)
 	assert.Len(t, scaledDeployments, 1)
 }
@@ -1153,3 +1143,126 @@ func TestScaleDownWithLock_FullCycle(t *testing.T) {
 	require.NoError(t, err)
 	assert.Len(t, scaledApps, 1)
 }
+
+// The restore label sits on the workload only; pods must be found through spec.selector.
+func TestScaleDown_WaitsForPodsSelectedByWorkload(t *testing.T) {
+	podLabels := map[string]string{"app": "receiver"}
+	deploy := createDeployment("receiver", podLabels, 1)
+	deploy.Labels = map[string]string{"restore": "true"}
+	sts := createStatefulSet("receiver-sts", podLabels, 1)
+	sts.Labels = map[string]string{"restore": "true"}
+	sts.Spec.Selector = &metav1.LabelSelector{MatchLabels: map[string]string{"app": "receiver-sts"}}
+
+	tests := []struct {
+		name string
+		kind string
+		pod  string
+	}{
+		{"deployment", "deployment", "receiver-0"},
+		{"statefulset", "statefulset", "receiver-sts-0"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			fakeClient := fake.NewSimpleClientset()
+			client := k8s.NewTestClient(fakeClient)
+			ctx := context.Background()
+			podAppLabel := "receiver"
+			if tt.kind == "deployment" {
+				_, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{})
+				require.NoError(t, err)
+			} else {
+				podAppLabel = "receiver-sts"
+				_, err := fakeClient.AppsV1().StatefulSets("test-ns").Create(ctx, &sts, metav1.CreateOptions{})
+				require.NoError(t, err)
+			}
+			pod := createPod(tt.pod, map[string]string{"app": podAppLabel}, corev1.PodRunning)
+			_, err := fakeClient.CoreV1().Pods("test-ns").Create(ctx, &pod, metav1.CreateOptions{})
+			require.NoError(t, err)
+
+			go func() {
+				time.Sleep(3 * time.Second)
+				_ = fakeClient.CoreV1().Pods("test-ns").Delete(ctx, tt.pod, metav1.DeleteOptions{})
+			}()
+
+			_, err = ScaleDown(client, "test-ns", "restore=true", logger.New(true, false))
+			require.NoError(t, err)
+
+			_, err = fakeClient.CoreV1().Pods("test-ns").Get(ctx, tt.pod, metav1.GetOptions{})
+			assert.True(t, apierrors.IsNotFound(err), "ScaleDown returned while the pod still existed")
+		})
+	}
+}
+
+func TestScaleDown_IgnoresFinishedPodsButWaitsForRunningOnes(t *testing.T) {
+	fakeClient := fake.NewSimpleClientset()
+	client := k8s.NewTestClient(fakeClient)
+	ctx := context.Background()
+	labels := map[string]string{"app": "test"}
+
+	deploy := createDeployment("deploy1", labels, 1)
+	_, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{})
+	require.NoError(t, err)
+	for _, pod := range []corev1.Pod{
+		createPod("evicted", labels, corev1.PodFailed),
+		createPod("done", labels, corev1.PodSucceeded),
+		createPod("running", labels, corev1.PodRunning),
+	} {
+		_, err = fakeClient.CoreV1().Pods("test-ns").Create(ctx, &pod, metav1.CreateOptions{})
+		require.NoError(t, err)
+	}
+
+	go func() {
+		time.Sleep(3 * time.Second)
+		_ = fakeClient.CoreV1().Pods("test-ns").Delete(ctx, "running", metav1.DeleteOptions{})
+	}()
+
+	_, err = ScaleDown(client, "test-ns", "app=test", logger.New(true, false))
+	require.NoError(t, err)
+
+	_, err = fakeClient.CoreV1().Pods("test-ns").Get(ctx, "running", metav1.GetOptions{})
+	assert.True(t, apierrors.IsNotFound(err), "ScaleDown returned while the running pod still existed")
+	_, err = fakeClient.CoreV1().Pods("test-ns").Get(ctx, "evicted", metav1.GetOptions{})
+	assert.NoError(t, err)
+}
+
+// A retry after a failed scale-down finds the workload already at 0 and must keep the recorded count.
+func TestScaleDown_RetryKeepsRecordedReplicas(t *testing.T) {
+	fakeClient := fake.NewSimpleClientset()
+	client := k8s.NewTestClient(fakeClient)
+	ctx := context.Background()
+	labels := map[string]string{"app": "test"}
+
+	deploy := createDeployment("deploy1", labels, 3)
+	_, err := fakeClient.AppsV1().Deployments("test-ns").Create(ctx, &deploy, metav1.CreateOptions{})
+	require.NoError(t, err)
+	sts := createStatefulSet("sts1", labels, 2)
+	_, err = fakeClient.AppsV1().StatefulSets("test-ns").Create(ctx, &sts, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	for range 2 {
+		scaled, err := ScaleDown(client, "test-ns", "app=test", logger.New(true, false))
+		require.NoError(t, err)
+		require.Len(t, scaled, 2)
+		assert.Equal(t, int32(3), scaled[0].Replicas)
+		assert.Equal(t, int32(2), scaled[1].Replicas)
+	}
+
+	deployAfter, err := fakeClient.AppsV1().Deployments("test-ns").Get(ctx, "deploy1", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, "3", deployAfter.Annotations[k8s.PreRestoreReplicasAnnotation])
+	stsAfter, err := fakeClient.AppsV1().StatefulSets("test-ns").Get(ctx, "sts1", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, "2", stsAfter.Annotations[k8s.PreRestoreReplicasAnnotation])
+
+	_, err = client.ScaleUpDeploymentsFromAnnotations("test-ns", "app=test")
+	require.NoError(t, err)
+	_, err = client.ScaleUpStatefulSetsFromAnnotations("test-ns", "app=test")
+	require.NoError(t, err)
+
+	deployUp, err := fakeClient.AppsV1().Deployments("test-ns").Get(ctx, "deploy1", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, int32(3), *deployUp.Spec.Replicas)
+	stsUp, err := fakeClient.AppsV1().StatefulSets("test-ns").Get(ctx, "sts1", metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, int32(2), *stsUp.Spec.Replicas)
+}