Bind migration requests and prepared exports

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/api/DeploymentApiContract.java

@@ -196,18 +196,22 @@ public final class DeploymentApiContract {
 
     /** H2-to-external-database migration input. */
     public record MetadataMigrationRequest(
+            @NotBlank String operationId,
             @NotNull MigrationTarget target,
             @NotNull @Valid MetadataDatabaseConfiguration targetDatabase,
             @NotNull ApplyMode applyMode) {
 
         public MetadataMigrationRequest {
+            if (!OperationIdValidator.isSafe(operationId)) {
+                throw new IllegalArgumentException("Invalid migration operation id");
+            }
             MigrationContractValidator.validateTarget(target, targetDatabase);
         }
 
         @Override
         public String toString() {
-            return "MetadataMigrationRequest[target=" + target + ", targetDatabase=<redacted>, applyMode="
-                    + applyMode + "]";
+            return "MetadataMigrationRequest[operationId=" + operationId + ", target=" + target
+                    + ", targetDatabase=<redacted>, applyMode=" + applyMode + "]";
         }
     }
 

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/api/DeploymentController.java

@@ -17,7 +17,9 @@
 
 package org.apache.hertzbeat.manager.setup.api;
 
+import jakarta.servlet.http.HttpServletResponse;
 import jakarta.validation.Valid;
+import java.io.IOException;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.ActivateMigrationRequest;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.DeploymentView;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MetadataMigrationRequest;
@@ -26,9 +28,9 @@ import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationExp
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationView;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ValidationResponse;
-import org.apache.hertzbeat.manager.setup.workflow.MigrationExportRenderer;
+import org.apache.hertzbeat.manager.setup.workflow.PreparedMigrationExport;
+import org.apache.hertzbeat.manager.setup.workflow.StagedMigrationExport;
 import org.springframework.beans.factory.ObjectProvider;
-import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -36,20 +38,15 @@ import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RestController;
-import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
 
 /** Transport-only adapter that fails safely until the migration workflow is available. */
 @RestController
 public final class DeploymentController {
 
     private final ObjectProvider<DeploymentWorkflow> workflowProvider;
-    private final ObjectProvider<MigrationExportRenderer> rendererProvider;
 
-    public DeploymentController(
-            ObjectProvider<DeploymentWorkflow> workflowProvider,
-            ObjectProvider<MigrationExportRenderer> rendererProvider) {
+    public DeploymentController(ObjectProvider<DeploymentWorkflow> workflowProvider) {
         this.workflowProvider = workflowProvider;
-        this.rendererProvider = rendererProvider;
     }
 
     @GetMapping(DeploymentApiContract.DEPLOYMENT_PATH)
@@ -86,16 +83,25 @@ public final class DeploymentController {
     }
 
     @PostMapping(DeploymentApiContract.EXPORT_PATH)
-    public ResponseEntity<StreamingResponseBody> export(
-            @PathVariable String operationId, @Valid @RequestBody MigrationExportRequest request) {
+    public void export(
+            @PathVariable String operationId, @Valid @RequestBody MigrationExportRequest request,
+            HttpServletResponse response) throws IOException {
         requireOperationId(operationId);
-        MigrationExportRenderer renderer = renderer();
-        ExportResponse metadata = workflow().prepareExport(operationId, request);
-        StreamingResponseBody body = output -> renderer.write(operationId, request, output);
-        return SetupHttpContract.noStore()
-                .header(HttpHeaders.CONTENT_DISPOSITION,
-                        "attachment; filename=\"" + metadata.fileName() + "\"")
-                .header(HttpHeaders.CONTENT_TYPE, metadata.mediaType()).body(body);
+        boolean responseMutated = false;
+        try (PreparedMigrationExport prepared = workflow().prepareExport(operationId, request);
+                StagedMigrationExport staged = prepared.stage()) {
+            ExportResponse metadata = staged.metadata();
+            responseMutated = true;
+            response.setHeader("Cache-Control", "no-store");
+            response.setHeader("Content-Disposition", "attachment; filename=\"" + metadata.fileName() + "\"");
+            response.setContentType(metadata.mediaType());
+            response.setContentLength(staged.size());
+            staged.writeTo(response.getOutputStream());
+            response.flushBuffer();
+        } catch (IOException | RuntimeException failure) {
+            resetUncommitted(response, responseMutated);
+            throw failure;
+        }
     }
 
     private DeploymentWorkflow workflow() {
@@ -112,16 +118,22 @@ public final class DeploymentController {
         }
     }
 
-    private MigrationExportRenderer renderer() {
-        MigrationExportRenderer renderer = rendererProvider.getIfUnique();
-        if (renderer == null) {
-            throw unavailable();
-        }
-        return renderer;
-    }
-
     private SetupApiException unavailable() {
         return new SetupApiException(
                 SetupApiContract.SetupErrorCode.MIGRATION_UNAVAILABLE, HttpStatus.SERVICE_UNAVAILABLE);
     }
+
+    private void resetUncommitted(HttpServletResponse response, boolean responseMutated) {
+        if (!responseMutated) {
+            return;
+        }
+        try {
+            if (!response.isCommitted()) {
+                response.reset();
+            }
+        } catch (RuntimeException ignored) {
+            // Preserve the original safe transport failure; response rollback is best effort.
+        }
+    }
+
 }

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/api/DeploymentWorkflow.java

@@ -23,8 +23,8 @@ import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MetadataMigr
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MetadataMigrationValidationRequest;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationExportRequest;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationView;
-import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ValidationResponse;
+import org.apache.hertzbeat.manager.setup.workflow.PreparedMigrationExport;
 
 /** Authenticated deployment boundary implemented by a later migration engine milestone. */
 public interface DeploymentWorkflow {
@@ -39,5 +39,5 @@ public interface DeploymentWorkflow {
 
     MigrationView activate(String operationId, ActivateMigrationRequest request);
 
-    ExportResponse prepareExport(String operationId, MigrationExportRequest request);
+    PreparedMigrationExport prepareExport(String operationId, MigrationExportRequest request);
 }

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/BoundedMigrationExportBuffer.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Objects;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
+
+/** Bounded in-memory staging buffer whose backing bytes have explicit ownership. */
+final class BoundedMigrationExportBuffer extends OutputStream implements AutoCloseable {
+
+    private static final int INITIAL_CAPACITY = 1024;
+
+    private final int limit;
+    private final PayloadFactory payloadFactory;
+    private byte[] bytes;
+    private int size;
+    private boolean transferred;
+    private boolean closed;
+
+    BoundedMigrationExportBuffer(int limit) {
+        this(limit, StagedMigrationExport::new);
+    }
+
+    BoundedMigrationExportBuffer(int limit, PayloadFactory payloadFactory) {
+        if (limit < 1) {
+            throw new IllegalArgumentException("limit");
+        }
+        this.limit = limit;
+        this.payloadFactory = Objects.requireNonNull(payloadFactory, "payloadFactory");
+        bytes = new byte[Math.min(limit, INITIAL_CAPACITY)];
+    }
+
+    @Override
+    public void write(int value) throws IOException {
+        requireCapacity(1);
+        bytes[size++] = (byte) value;
+    }
+
+    @Override
+    public void write(byte[] source, int offset, int length) throws IOException {
+        Objects.checkFromIndexSize(offset, length, source.length);
+        requireCapacity(length);
+        System.arraycopy(source, offset, bytes, size, length);
+        size += length;
+    }
+
+    StagedMigrationExport finish(ExportResponse metadata) {
+        if (transferred || closed) {
+            throw new PreparedMigrationExportException();
+        }
+        try {
+            StagedMigrationExport staged = payloadFactory.create(metadata, bytes, size);
+            transferred = true;
+            bytes = new byte[0];
+            return staged;
+        } catch (RuntimeException | Error failure) {
+            close();
+            throw failure;
+        }
+    }
+
+    @Override
+    public void close() {
+        if (closed) {
+            return;
+        }
+        Arrays.fill(bytes, (byte) 0);
+        bytes = new byte[0];
+        size = 0;
+        closed = true;
+    }
+
+    private void requireCapacity(int additional) throws IOException {
+        if (transferred || closed || additional > limit - size) {
+            throw new IOException("Prepared migration export exceeded its size limit");
+        }
+        int required = size + additional;
+        if (required <= bytes.length) {
+            return;
+        }
+        int nextCapacity = Math.min(limit, Math.max(required, bytes.length * 2));
+        byte[] next = Arrays.copyOf(bytes, nextCapacity);
+        Arrays.fill(bytes, (byte) 0);
+        bytes = next;
+    }
+
+    @FunctionalInterface
+    interface PayloadFactory {
+        StagedMigrationExport create(ExportResponse metadata, byte[] bytes, int size);
+    }
+}

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/MigrationExportRenderer.java

@@ -1,29 +1,19 @@
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
+ * contributor license agreements. See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
  */
 
 package org.apache.hertzbeat.manager.setup.workflow;
 
 import java.io.IOException;
 import java.io.OutputStream;
-import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationExportRequest;
+import org.apache.hertzbeat.manager.setup.config.SecretValue;
 
-/** Streaming port for one-shot external-apply content; implementations must not retain request secrets. */
+/** Pure bound renderer with no owned resources; the borrowed secret must not be retained. */
 @FunctionalInterface
 public interface MigrationExportRenderer {
 
-    void write(String operationId, MigrationExportRequest request, OutputStream output) throws IOException;
+    void write(SecretValue borrowedSecret, OutputStream output) throws IOException;
 }

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/NonClosingMigrationExportStream.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Objects;
+
+/** Renderer-facing view that cannot close or clear its owner-managed buffer. */
+final class NonClosingMigrationExportStream extends OutputStream {
+
+    private final OutputStream delegate;
+
+    NonClosingMigrationExportStream(OutputStream delegate) {
+        this.delegate = Objects.requireNonNull(delegate, "delegate");
+    }
+
+    @Override
+    public void write(int value) throws IOException {
+        delegate.write(value);
+    }
+
+    @Override
+    public void write(byte[] bytes, int offset, int length) throws IOException {
+        delegate.write(bytes, offset, length);
+    }
+
+    @Override
+    public void flush() throws IOException {
+        delegate.flush();
+    }
+
+    @Override
+    public void close() {
+        // The renderer borrows this view; buffer ownership remains with PreparedMigrationExport.
+    }
+}

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/PreparedMigrationExport.java

@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import java.io.IOException;
+import java.util.Objects;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
+import org.apache.hertzbeat.manager.setup.config.SecretValue;
+
+/** One-shot export preparation that owns only an independent clearable secret. */
+public final class PreparedMigrationExport implements AutoCloseable {
+
+    static final int MAX_EXPORT_BYTES = 1024 * 1024;
+
+    private final ExportResponse metadata;
+    private final SecretValue ownedSecret;
+    private final MigrationExportRenderer renderer;
+    private final BufferFactory bufferFactory;
+    private State state = State.READY;
+
+    private PreparedMigrationExport(
+            ExportResponse metadata, SecretValue ownedSecret, MigrationExportRenderer renderer,
+            BufferFactory bufferFactory) {
+        this.metadata = Objects.requireNonNull(metadata, "metadata");
+        this.ownedSecret = Objects.requireNonNull(ownedSecret, "ownedSecret");
+        this.renderer = Objects.requireNonNull(renderer, "renderer");
+        this.bufferFactory = Objects.requireNonNull(bufferFactory, "bufferFactory");
+    }
+
+    /** Copies the borrowed secret immediately; ownership of the caller's value never changes. */
+    static PreparedMigrationExport prepare(
+            ExportResponse metadata, SecretValue borrowedSecret, MigrationExportRenderer renderer) {
+        Objects.requireNonNull(metadata, "metadata");
+        Objects.requireNonNull(borrowedSecret, "borrowedSecret");
+        Objects.requireNonNull(renderer, "renderer");
+        return prepare(metadata, borrowedSecret, renderer, BoundedMigrationExportBuffer::new);
+    }
+
+    static PreparedMigrationExport prepare(
+            ExportResponse metadata, SecretValue borrowedSecret, MigrationExportRenderer renderer,
+            BufferFactory bufferFactory) {
+        Objects.requireNonNull(metadata, "metadata");
+        Objects.requireNonNull(borrowedSecret, "borrowedSecret");
+        Objects.requireNonNull(renderer, "renderer");
+        Objects.requireNonNull(bufferFactory, "bufferFactory");
+        return new PreparedMigrationExport(
+                metadata, SecretValue.copyOf(borrowedSecret), renderer, bufferFactory);
+    }
+
+    public ExportResponse metadata() {
+        return metadata;
+    }
+
+    /** Renders and finalizes a bounded payload before any HTTP response is mutated. */
+    public StagedMigrationExport stage() throws IOException {
+        beginStage();
+        BoundedMigrationExportBuffer buffer = null;
+        boolean transferred = false;
+        try {
+            buffer = bufferFactory.create(MAX_EXPORT_BYTES);
+            renderer.write(ownedSecret, new NonClosingMigrationExportStream(buffer));
+            StagedMigrationExport staged = buffer.finish(metadata);
+            transferred = true;
+            return staged;
+        } catch (IOException failure) {
+            throw new IOException("Prepared migration export failed");
+        } catch (RuntimeException failure) {
+            throw new PreparedMigrationExportException();
+        } finally {
+            if (!transferred && buffer != null) {
+                buffer.close();
+            }
+            ownedSecret.close();
+            finishStage();
+        }
+    }
+
+    @Override
+    public synchronized void close() {
+        if (state == State.CLOSED) {
+            return;
+        }
+        if (state == State.STAGING) {
+            throw new PreparedMigrationExportException();
+        }
+        state = State.CLOSED;
+        ownedSecret.close();
+    }
+
+    @Override
+    public synchronized String toString() {
+        return "PreparedMigrationExport[state=" + state + "]";
+    }
+
+    private synchronized void beginStage() {
+        if (state != State.READY) {
+            throw new PreparedMigrationExportException();
+        }
+        state = State.STAGING;
+    }
+
+    private synchronized void finishStage() {
+        state = State.CLOSED;
+    }
+
+    private enum State {
+        READY,
+        STAGING,
+        CLOSED
+    }
+
+    @FunctionalInterface
+    interface BufferFactory {
+        BoundedMigrationExportBuffer create(int limit);
+    }
+}

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/PreparedMigrationExportException.java

@@ -0,0 +1,16 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+/** Cause-free lifecycle failure for a prepared export capability. */
+final class PreparedMigrationExportException extends IllegalStateException {
+
+    PreparedMigrationExportException() {
+        super("Prepared migration export is unavailable");
+    }
+}

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/setup/workflow/StagedMigrationExport.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Objects;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
+
+/** Finalized bounded export payload; closing clears every owned byte. */
+public final class StagedMigrationExport implements AutoCloseable {
+
+    private final ExportResponse metadata;
+    private byte[] bytes;
+    private final int size;
+    private State state = State.READY;
+
+    StagedMigrationExport(ExportResponse metadata, byte[] bytes, int size) {
+        this.metadata = Objects.requireNonNull(metadata, "metadata");
+        this.bytes = Objects.requireNonNull(bytes, "bytes");
+        if (size < 0 || size > bytes.length) {
+            throw new IllegalArgumentException("size");
+        }
+        this.size = size;
+    }
+
+    public ExportResponse metadata() {
+        return metadata;
+    }
+
+    public int size() {
+        return size;
+    }
+
+    public void writeTo(OutputStream output) throws IOException {
+        Objects.requireNonNull(output, "output");
+        byte[] payload = beginWrite();
+        try {
+            output.write(payload, 0, size);
+        } finally {
+            finishWrite();
+        }
+    }
+
+    @Override
+    public synchronized void close() {
+        if (state == State.CLOSED) {
+            return;
+        }
+        if (state == State.WRITING) {
+            throw new PreparedMigrationExportException();
+        }
+        clear();
+    }
+
+    private synchronized byte[] beginWrite() {
+        if (state != State.READY) {
+            throw new PreparedMigrationExportException();
+        }
+        state = State.WRITING;
+        return bytes;
+    }
+
+    private synchronized void finishWrite() {
+        clear();
+    }
+
+    private void clear() {
+        Arrays.fill(bytes, (byte) 0);
+        bytes = new byte[0];
+        state = State.CLOSED;
+    }
+
+    private enum State {
+        READY,
+        WRITING,
+        CLOSED
+    }
+}

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/api/DeploymentApiContractTest.java

@@ -45,6 +45,7 @@ import org.apache.hertzbeat.manager.setup.api.SetupApiContract.MetadataDatabaseK
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.SetupErrorCode;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.TelemetryStoreKind;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.TelemetryStoreSummary;
+import org.apache.hertzbeat.manager.setup.workflow.PreparedMigrationExport;
 import org.junit.jupiter.api.Test;
 
 /** Freezes authenticated deployment and H2 migration contracts. */
@@ -69,7 +70,8 @@ class DeploymentApiContractTest {
                 "maintenanceAdmission", "activeOperationId");
         assertComponents(DeploymentApiContract.MetadataMigrationValidationRequest.class,
                 "target", "targetDatabase");
-        assertComponents(DeploymentApiContract.MetadataMigrationRequest.class, "target", "targetDatabase", "applyMode");
+        assertComponents(DeploymentApiContract.MetadataMigrationRequest.class,
+                "operationId", "target", "targetDatabase", "applyMode");
         assertComponents(DeploymentApiContract.MigrationView.class, "operationId", "state", "source", "target",
                 "stage", "progressPercent", "createdAt", "startedAt", "completedAt", "verificationState",
                 "errorCode", "nextPollAfterMillis", "activationAvailable", "restartRequired",
@@ -96,6 +98,10 @@ class DeploymentApiContractTest {
                 DeploymentWorkflow.class.getMethod(
                         "validate", DeploymentApiContract.MetadataMigrationValidationRequest.class)
                         .getReturnType());
+        assertEquals(PreparedMigrationExport.class,
+                DeploymentWorkflow.class.getMethod(
+                        "prepareExport", String.class, DeploymentApiContract.MigrationExportRequest.class)
+                        .getReturnType());
     }
 
     @Test
@@ -183,15 +189,22 @@ class DeploymentApiContractTest {
         MetadataDatabaseConfiguration mysql = new MetadataDatabaseConfiguration(
                 MetadataDatabaseKind.MYSQL, "jdbc:mysql://db/hertzbeat", "user", "secret");
         assertEquals(mysql, new DeploymentApiContract.MetadataMigrationRequest(
-                MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE).targetDatabase());
+                "migration-1", MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE).targetDatabase());
         MetadataDatabaseConfiguration postgres = new MetadataDatabaseConfiguration(
                 MetadataDatabaseKind.POSTGRESQL, "jdbc:postgresql://db/hertzbeat", "user", "secret");
         assertThrows(IllegalArgumentException.class, () -> new DeploymentApiContract.MetadataMigrationRequest(
-                MigrationTarget.MYSQL, postgres, ApplyMode.MANAGED_WRITE));
+                "migration-1", MigrationTarget.MYSQL, postgres, ApplyMode.MANAGED_WRITE));
         MetadataDatabaseConfiguration h2 = new MetadataDatabaseConfiguration(
                 MetadataDatabaseKind.H2, "jdbc:h2:file:./data/hertzbeat", "sa", "secret");
         assertThrows(IllegalArgumentException.class, () -> new DeploymentApiContract.MetadataMigrationRequest(
-                MigrationTarget.POSTGRESQL, h2, ApplyMode.EXTERNAL_APPLY));
+                "migration-1", MigrationTarget.POSTGRESQL, h2, ApplyMode.EXTERNAL_APPLY));
+
+        assertThrows(IllegalArgumentException.class, () -> new DeploymentApiContract.MetadataMigrationRequest(
+                null, MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE));
+        assertThrows(IllegalArgumentException.class, () -> new DeploymentApiContract.MetadataMigrationRequest(
+                ".hidden", MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE));
+        assertThrows(IllegalArgumentException.class, () -> new DeploymentApiContract.MetadataMigrationRequest(
+                "a".repeat(129), MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE));
     }
 
     @Test
@@ -200,13 +213,17 @@ class DeploymentApiContractTest {
         MetadataDatabaseConfiguration mysql = new MetadataDatabaseConfiguration(
                 MetadataDatabaseKind.MYSQL, "jdbc:mysql://db/hertzbeat", "user", secret);
         DeploymentApiContract.MetadataMigrationRequest request = new DeploymentApiContract.MetadataMigrationRequest(
-                MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE);
+                "migration-1", MigrationTarget.MYSQL, mysql, ApplyMode.MANAGED_WRITE);
         assertFalse(objectMapper.writeValueAsString(request).contains(secret));
         assertFalse(request.toString().contains(secret));
         DeploymentApiContract.MigrationExportRequest export = new DeploymentApiContract.MigrationExportRequest(
                 SetupApiContract.ExportFormat.ENV, MigrationOperationState.AWAITING_EXTERNAL_APPLY, mysql);
         assertFalse(objectMapper.writeValueAsString(export).contains(secret));
         assertFalse(export.toString().contains(secret));
+        assertThrows(IllegalArgumentException.class,
+                () -> new SetupApiContract.ExportResponse("unsafe\".env", "text/plain"));
+        assertThrows(IllegalArgumentException.class,
+                () -> new SetupApiContract.ExportResponse("safe.env", "text/plain\r\nprivate-header"));
     }
 
     @Test

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/api/DeploymentControllerExportTest.java

@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.api;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import jakarta.servlet.ServletOutputStream;
+import jakarta.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.util.List;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationExportRequest;
+import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationOperationState;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportFormat;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.MetadataDatabaseConfiguration;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.MetadataDatabaseKind;
+import org.apache.hertzbeat.manager.setup.workflow.PreparedMigrationExport;
+import org.apache.hertzbeat.manager.setup.workflow.StagedMigrationExport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.support.StaticListableBeanFactory;
+import org.springframework.http.MediaType;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+/** Bounded synchronous export and servlet failure contracts for deployment routes. */
+class DeploymentControllerExportTest {
+
+    private final DeploymentWorkflow workflow = mock(DeploymentWorkflow.class);
+    private ObjectProvider<DeploymentWorkflow> workflowProvider;
+    private MockMvc mvc;
+
+    @BeforeEach
+    void setUp() {
+        StaticListableBeanFactory factory = new StaticListableBeanFactory();
+        factory.addBean("workflow", workflow);
+        workflowProvider = factory.getBeanProvider(DeploymentWorkflow.class);
+        mvc = MockMvcBuilders.standaloneSetup(new DeploymentController(workflowProvider))
+                .setControllerAdvice(new SetupExceptionHandler()).build();
+    }
+
+    @Test
+    void externalApplyExportWritesOnlyAfterNoStoreAttachmentIsPrepared() throws Exception {
+        ExportFixture fixture = fixture();
+        when(workflow.prepareExport(eq("migration-1"), any())).thenReturn(fixture.prepared);
+        String request = """
+                {"format":"env","expectedState":"awaiting_external_apply",
+                "targetDatabase":{"kind":"mysql","jdbcUrl":"jdbc:mysql://db/hertzbeat",
+                "username":"operator","password":"export-secret"}}
+                """;
+
+        mvc.perform(post(DeploymentApiContract.EXPORT_PATH, "migration-1")
+                        .contentType(MediaType.APPLICATION_JSON).content(request))
+                .andExpect(status().isOk())
+                .andExpect(header().string("Cache-Control", "no-store"))
+                .andExpect(header().string("Content-Disposition",
+                        "attachment; filename=\"hertzbeat-migration.env\""));
+
+        verify(fixture.staged).writeTo(any());
+        verify(fixture.staged).close();
+        verify(fixture.prepared).close();
+    }
+
+    @Test
+    void exportBodyIsWrittenSynchronouslyWithoutCapturingTheRequest() throws Exception {
+        ExportFixture fixture = fixture();
+
+        fixture.controller.export("migration-1", fixture.request, fixture.response);
+
+        verify(fixture.prepared).stage();
+        verify(fixture.staged).metadata();
+        verify(fixture.staged).writeTo(fixture.output);
+        verify(fixture.response).flushBuffer();
+        verify(fixture.staged).close();
+        verify(fixture.prepared).close();
+    }
+
+    @Test
+    void renderFailureClosesPreparedWithoutMutatingTheServletResponse() throws Exception {
+        MigrationExportRequest request = exportRequest();
+        PreparedMigrationExport prepared = mock(PreparedMigrationExport.class);
+        when(workflow.prepareExport("migration-1", request)).thenReturn(prepared);
+        when(prepared.stage()).thenThrow(new IOException("Prepared migration export failed"));
+        HttpServletResponse response = mock(HttpServletResponse.class);
+        DeploymentController controller = new DeploymentController(workflowProvider);
+
+        assertThatThrownBy(() -> controller.export("migration-1", request, response))
+                .isInstanceOf(IOException.class)
+                .hasMessageNotContaining("export-secret");
+
+        verify(prepared).close();
+        verifyNoInteractions(response);
+    }
+
+    @Test
+    void outputFailureClosesPreparedExportAndResetsAnUncommittedResponse() throws Exception {
+        ExportFixture fixture = fixture();
+        IOException outputFailure = new IOException("private-output-detail");
+        when(fixture.response.isCommitted()).thenReturn(false);
+        org.mockito.Mockito.doThrow(outputFailure).when(fixture.staged).writeTo(fixture.output);
+
+        assertThatThrownBy(() -> fixture.controller.export(
+                "migration-1", fixture.request, fixture.response)).isSameAs(outputFailure);
+
+        verify(fixture.response).reset();
+        verify(fixture.staged).close();
+        verify(fixture.prepared).close();
+    }
+
+    @Test
+    void committedTransportFailureDoesNotAttemptToResetTheResponse() throws Exception {
+        ExportFixture fixture = fixture();
+        IOException outputFailure = new IOException("private-output-detail");
+        when(fixture.response.isCommitted()).thenReturn(true);
+        org.mockito.Mockito.doThrow(outputFailure).when(fixture.staged).writeTo(fixture.output);
+
+        assertThatThrownBy(() -> fixture.controller.export(
+                "migration-1", fixture.request, fixture.response)).isSameAs(outputFailure);
+
+        verify(fixture.response, never()).reset();
+        verify(fixture.staged).close();
+        verify(fixture.prepared).close();
+    }
+
+    @Test
+    void ordinaryResponseRuntimeAtEveryMutationBoundaryResetsWithoutReplacingFailure() throws Exception {
+        for (ResponseFailurePoint failurePoint : ResponseFailurePoint.values()) {
+            ExportFixture fixture = fixture();
+            when(fixture.response.isCommitted()).thenReturn(false);
+            IllegalStateException failure = new IllegalStateException("private-response-runtime");
+            failurePoint.fail(fixture, failure);
+
+            assertThatThrownBy(() -> fixture.controller.export(
+                    "migration-1", fixture.request, fixture.response)).isSameAs(failure);
+
+            verify(fixture.response).reset();
+            verify(fixture.staged).close();
+            verify(fixture.prepared).close();
+        }
+    }
+
+    @Test
+    void rollbackRuntimeNeverReplacesTheOriginalResponseFailure() throws Exception {
+        for (boolean failCommittedCheck : List.of(true, false)) {
+            ExportFixture fixture = fixture();
+            IllegalStateException original = new IllegalStateException("original-response-failure");
+            org.mockito.Mockito.doThrow(original).when(fixture.staged).writeTo(fixture.output);
+            if (failCommittedCheck) {
+                when(fixture.response.isCommitted())
+                        .thenThrow(new IllegalStateException("private-committed-failure"));
+            } else {
+                when(fixture.response.isCommitted()).thenReturn(false);
+                org.mockito.Mockito.doThrow(new IllegalStateException("private-reset-failure"))
+                        .when(fixture.response).reset();
+            }
+
+            assertThatThrownBy(() -> fixture.controller.export(
+                    "migration-1", fixture.request, fixture.response)).isSameAs(original);
+        }
+    }
+
+    @Test
+    void fatalResponseErrorRemainsRawAndPrimary() throws Exception {
+        ExportFixture fixture = fixture();
+        AssertionError fatal = new AssertionError("response-fatal");
+        org.mockito.Mockito.doThrow(fatal).when(fixture.response)
+                .setHeader("Cache-Control", "no-store");
+
+        assertThatThrownBy(() -> fixture.controller.export(
+                "migration-1", fixture.request, fixture.response)).isSameAs(fatal);
+
+        verify(fixture.response, never()).isCommitted();
+        verify(fixture.response, never()).reset();
+        verify(fixture.staged).close();
+        verify(fixture.prepared).close();
+    }
+
+    private ExportFixture fixture() throws Exception {
+        MigrationExportRequest request = exportRequest();
+        PreparedMigrationExport prepared = mock(PreparedMigrationExport.class);
+        StagedMigrationExport staged = mock(StagedMigrationExport.class);
+        HttpServletResponse response = mock(HttpServletResponse.class);
+        ServletOutputStream output = mock(ServletOutputStream.class);
+        when(prepared.stage()).thenReturn(staged);
+        when(staged.metadata()).thenReturn(
+                new ExportResponse("hertzbeat-migration.env", "text/plain"));
+        when(staged.size()).thenReturn(8);
+        when(workflow.prepareExport("migration-1", request)).thenReturn(prepared);
+        when(response.getOutputStream()).thenReturn(output);
+        return new ExportFixture(request, prepared, staged, response, output,
+                new DeploymentController(workflowProvider));
+    }
+
+    private MigrationExportRequest exportRequest() {
+        return new MigrationExportRequest(ExportFormat.ENV,
+                MigrationOperationState.AWAITING_EXTERNAL_APPLY,
+                new MetadataDatabaseConfiguration(MetadataDatabaseKind.MYSQL,
+                        "jdbc:mysql://db/hertzbeat", "operator", "export-secret"));
+    }
+
+    private record ExportFixture(
+            MigrationExportRequest request,
+            PreparedMigrationExport prepared,
+            StagedMigrationExport staged,
+            HttpServletResponse response,
+            ServletOutputStream output,
+            DeploymentController controller) { }
+
+    private enum ResponseFailurePoint {
+        HEADER {
+            @Override
+            void fail(ExportFixture fixture, RuntimeException failure) {
+                org.mockito.Mockito.doThrow(failure).when(fixture.response)
+                        .setHeader("Cache-Control", "no-store");
+            }
+        },
+        OUTPUT_STREAM {
+            @Override
+            void fail(ExportFixture fixture, RuntimeException failure) throws IOException {
+                when(fixture.response.getOutputStream()).thenThrow(failure);
+            }
+        },
+        WRITE {
+            @Override
+            void fail(ExportFixture fixture, RuntimeException failure) throws IOException {
+                org.mockito.Mockito.doThrow(failure).when(fixture.staged).writeTo(fixture.output);
+            }
+        };
+
+        abstract void fail(ExportFixture fixture, RuntimeException failure) throws IOException;
+    }
+}

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/api/DeploymentControllerRegistrationTest.java

@@ -22,7 +22,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 
-import org.apache.hertzbeat.manager.setup.workflow.MigrationExportRenderer;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.ComponentScan;
@@ -48,9 +47,8 @@ class DeploymentControllerRegistrationTest {
     }
 
     @Test
-    void componentScanResolvesWorkflowAndRendererRegisteredAlongsideController() {
+    void componentScanResolvesWorkflowRegisteredAlongsideController() {
         context.withBean(DeploymentWorkflow.class, () -> mock(DeploymentWorkflow.class))
-                .withBean(MigrationExportRenderer.class, () -> mock(MigrationExportRenderer.class))
                 .run(result -> assertThat(result).hasSingleBean(DeploymentController.class));
     }
 

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/api/DeploymentControllerTest.java

@@ -18,21 +18,19 @@
 package org.apache.hertzbeat.manager.setup.api;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
-import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
-import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
 
-import java.io.ByteArrayOutputStream;
 import java.time.Instant;
 import java.util.List;
 import org.apache.hertzbeat.common.transaction.MetadataWriteAdmissionException;
@@ -41,49 +39,38 @@ import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.DeploymentVi
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MaintenanceAdmission;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MaintenanceMode;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationCapability;
-import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationExportRequest;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationOperationState;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationStage;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationTarget;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.MigrationView;
 import org.apache.hertzbeat.manager.setup.api.DeploymentApiContract.VerificationState;
-import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportFormat;
-import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ApplyMode;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ConfigSource;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ManagementDatabaseSummary;
-import org.apache.hertzbeat.manager.setup.api.SetupApiContract.MetadataDatabaseConfiguration;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.MetadataDatabaseKind;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ValidationResponse;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.TelemetryStoreKind;
 import org.apache.hertzbeat.manager.setup.api.SetupApiContract.TelemetryStoreSummary;
-import org.apache.hertzbeat.manager.setup.workflow.MigrationExportRenderer;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.support.StaticListableBeanFactory;
 import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
 import org.springframework.test.web.servlet.MockMvc;
-import org.springframework.test.web.servlet.MvcResult;
 import org.springframework.test.web.servlet.setup.MockMvcBuilders;
-import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;
 
 /** Transport proof for authenticated, no-store deployment routes. */
 class DeploymentControllerTest {
 
     private final DeploymentWorkflow workflow = mock(DeploymentWorkflow.class);
-    private final MigrationExportRenderer exportRenderer = mock(MigrationExportRenderer.class);
     private ObjectProvider<DeploymentWorkflow> workflowProvider;
-    private ObjectProvider<MigrationExportRenderer> rendererProvider;
     private MockMvc mvc;
 
     @BeforeEach
     void setUp() {
-        StaticListableBeanFactory factory = providerFactory(List.of(workflow), List.of(exportRenderer));
+        StaticListableBeanFactory factory = providerFactory(List.of(workflow));
         workflowProvider = factory.getBeanProvider(DeploymentWorkflow.class);
-        rendererProvider = factory.getBeanProvider(MigrationExportRenderer.class);
-        mvc = mvc(workflowProvider, rendererProvider);
+        mvc = mvc(workflowProvider);
     }
 
     @Test
@@ -99,7 +86,7 @@ class DeploymentControllerTest {
                 "jdbcUrl":"jdbc:mysql://db/hertzbeat","username":"operator","password":"request-secret"}}
                 """;
         String migration = """
-                {"target":"mysql","targetDatabase":{"kind":"mysql",
+                {"operationId":"migration-1","target":"mysql","targetDatabase":{"kind":"mysql",
                 "jdbcUrl":"jdbc:mysql://db/hertzbeat","username":"operator","password":"request-secret"},
                 "applyMode":"managed_write"}
                 """;
@@ -130,6 +117,7 @@ class DeploymentControllerTest {
                 .andExpect(jsonPath("$.restartRequired").value(true));
 
         verify(workflow).migration("migration-1");
+        verify(workflow).migrate(argThat(request -> "migration-1".equals(request.operationId())));
     }
 
     @Test
@@ -171,7 +159,27 @@ class DeploymentControllerTest {
     }
 
     @Test
-    void rejectsInvalidOperationIdsBeforeWorkflowOrRendererDispatch() throws Exception {
+    void rejectsInvalidOperationIdsBeforeWorkflowDispatch() throws Exception {
+        String migrationWithoutId = """
+                {"target":"mysql","targetDatabase":{"kind":"mysql",
+                "jdbcUrl":"jdbc:mysql://db/hertzbeat","username":"operator","password":"request-secret"},
+                "applyMode":"managed_write"}
+                """;
+        String migrationWithUnsafeId = """
+                {"operationId":".hidden","target":"mysql","targetDatabase":{"kind":"mysql",
+                "jdbcUrl":"jdbc:mysql://db/hertzbeat","username":"operator","password":"request-secret"},
+                "applyMode":"managed_write"}
+                """;
+        mvc.perform(post(DeploymentApiContract.MIGRATION_PATH).contentType(MediaType.APPLICATION_JSON)
+                        .content(migrationWithoutId))
+                .andExpect(status().isBadRequest())
+                .andExpect(header().string("Cache-Control", "no-store"))
+                .andExpect(jsonPath("$.errorCode").value("invalid_request"));
+        mvc.perform(post(DeploymentApiContract.MIGRATION_PATH).contentType(MediaType.APPLICATION_JSON)
+                        .content(migrationWithUnsafeId))
+                .andExpect(status().isBadRequest())
+                .andExpect(header().string("Cache-Control", "no-store"))
+                .andExpect(jsonPath("$.errorCode").value("invalid_request"));
         mvc.perform(get(DeploymentApiContract.MIGRATION_OPERATION_PATH, ".hidden"))
                 .andExpect(status().isBadRequest())
                 .andExpect(header().string("Cache-Control", "no-store"))
@@ -193,63 +201,17 @@ class DeploymentControllerTest {
                 .andExpect(header().string("Cache-Control", "no-store"))
                 .andExpect(jsonPath("$.errorCode").value("invalid_request"));
 
-        verifyNoInteractions(workflow, exportRenderer);
+        verifyNoInteractions(workflow);
     }
 
     @Test
     void missingWorkflowReturnsStableNoStoreUnavailable() throws Exception {
-        assertDeploymentUnavailable(mvc(List.of(), List.of(exportRenderer)));
+        assertDeploymentUnavailable(mvc(List.of()));
     }
 
     @Test
     void ambiguousWorkflowReturnsStableNoStoreUnavailable() throws Exception {
-        assertDeploymentUnavailable(mvc(
-                List.of(workflow, mock(DeploymentWorkflow.class)), List.of(exportRenderer)));
-    }
-
-    @Test
-    void missingOrAmbiguousRendererReturnsStableNoStoreUnavailable() throws Exception {
-        assertExportUnavailable(mvc(List.of(workflow), List.of()));
-        assertExportUnavailable(mvc(List.of(workflow),
-                List.of(exportRenderer, mock(MigrationExportRenderer.class))));
-    }
-
-    @Test
-    void externalApplyExportStreamsOnlyAfterNoStoreAttachmentIsPrepared() throws Exception {
-        when(workflow.prepareExport(eq("migration-1"), any())).thenReturn(
-                new ExportResponse("hertzbeat-migration.env", "text/plain"));
-        String request = """
-                {"format":"env","expectedState":"awaiting_external_apply",
-                "targetDatabase":{"kind":"mysql","jdbcUrl":"jdbc:mysql://db/hertzbeat",
-                "username":"operator","password":"export-secret"}}
-                """;
-
-        MvcResult pending = mvc.perform(post(DeploymentApiContract.EXPORT_PATH, "migration-1")
-                        .contentType(MediaType.APPLICATION_JSON).content(request))
-                .andExpect(request().asyncStarted()).andReturn();
-        mvc.perform(asyncDispatch(pending))
-                .andExpect(status().isOk())
-                .andExpect(header().string("Cache-Control", "no-store"))
-                .andExpect(header().string("Content-Disposition",
-                        "attachment; filename=\"hertzbeat-migration.env\""));
-        verify(exportRenderer).write(eq("migration-1"), any(), any());
-    }
-
-    @Test
-    void exportBodyIsDeferredUntilTheStreamingCallbackRuns() throws Exception {
-        MigrationExportRequest request = new MigrationExportRequest(ExportFormat.ENV,
-                MigrationOperationState.AWAITING_EXTERNAL_APPLY,
-                new MetadataDatabaseConfiguration(MetadataDatabaseKind.MYSQL,
-                        "jdbc:mysql://db/hertzbeat", "operator", "export-secret"));
-        when(workflow.prepareExport("migration-1", request)).thenReturn(
-                new ExportResponse("hertzbeat-migration.env", "text/plain"));
-        DeploymentController controller = new DeploymentController(workflowProvider, rendererProvider);
-
-        ResponseEntity<StreamingResponseBody> response = controller.export("migration-1", request);
-
-        verifyNoInteractions(exportRenderer);
-        response.getBody().writeTo(new ByteArrayOutputStream());
-        verify(exportRenderer).write(eq("migration-1"), eq(request), any());
+        assertDeploymentUnavailable(mvc(List.of(workflow, mock(DeploymentWorkflow.class))));
     }
 
     private MigrationView readyMigration() {
@@ -277,29 +239,21 @@ class DeploymentControllerTest {
                 VerificationState.SUCCEEDED, null, 1000, false, true, false);
     }
 
-    private MockMvc mvc(
-            List<DeploymentWorkflow> workflows, List<MigrationExportRenderer> renderers) {
-        StaticListableBeanFactory factory = providerFactory(workflows, renderers);
-        return mvc(factory.getBeanProvider(DeploymentWorkflow.class),
-                factory.getBeanProvider(MigrationExportRenderer.class));
+    private MockMvc mvc(List<DeploymentWorkflow> workflows) {
+        StaticListableBeanFactory factory = providerFactory(workflows);
+        return mvc(factory.getBeanProvider(DeploymentWorkflow.class));
     }
 
-    private MockMvc mvc(
-            ObjectProvider<DeploymentWorkflow> workflows,
-            ObjectProvider<MigrationExportRenderer> renderers) {
-        return MockMvcBuilders.standaloneSetup(new DeploymentController(workflows, renderers))
+    private MockMvc mvc(ObjectProvider<DeploymentWorkflow> workflows) {
+        return MockMvcBuilders.standaloneSetup(new DeploymentController(workflows))
                 .setControllerAdvice(new SetupExceptionHandler()).build();
     }
 
-    private StaticListableBeanFactory providerFactory(
-            List<DeploymentWorkflow> workflows, List<MigrationExportRenderer> renderers) {
+    private StaticListableBeanFactory providerFactory(List<DeploymentWorkflow> workflows) {
         StaticListableBeanFactory factory = new StaticListableBeanFactory();
         for (int index = 0; index < workflows.size(); index++) {
             factory.addBean("workflow-" + index, workflows.get(index));
         }
-        for (int index = 0; index < renderers.size(); index++) {
-            factory.addBean("renderer-" + index, renderers.get(index));
-        }
         return factory;
     }
 
@@ -310,16 +264,4 @@ class DeploymentControllerTest {
                 .andExpect(jsonPath("$.errorCode").value("migration_unavailable"));
     }
 
-    private void assertExportUnavailable(MockMvc candidate) throws Exception {
-        String request = """
-                {"format":"env","expectedState":"awaiting_external_apply",
-                "targetDatabase":{"kind":"mysql","jdbcUrl":"jdbc:mysql://db/hertzbeat",
-                "username":"operator","password":"export-secret"}}
-                """;
-        candidate.perform(post(DeploymentApiContract.EXPORT_PATH, "migration-1")
-                        .contentType(MediaType.APPLICATION_JSON).content(request))
-                .andExpect(status().isServiceUnavailable())
-                .andExpect(header().string("Cache-Control", "no-store"))
-                .andExpect(jsonPath("$.errorCode").value("migration_unavailable"));
-    }
 }

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/api/SetupApiContractTest.java

@@ -222,8 +222,9 @@ class SetupApiContractTest {
                 "public_address_invalid", "mail_connection_failed", "administrator_already_configured",
                 "administrator_username_invalid", "operation_not_found", "operation_conflict",
                 "migration_source_unsupported", "migration_target_not_empty", "migration_multi_node_unsupported",
-                "migration_copy_failed", "migration_verification_failed", "migration_activation_failed",
-                "restart_failed");
+                "migration_topology_unavailable", "migration_maintenance_required", "migration_unavailable",
+                "migration_copy_failed", "migration_verification_failed", "migration_activation_not_available",
+                "migration_activation_failed", "restart_failed");
     }
 
     @Test

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/BoundedMigrationExportBufferTest.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
+import org.junit.jupiter.api.Test;
+
+/** Exact byte ownership contracts for finalized migration export staging. */
+class BoundedMigrationExportBufferTest {
+
+    @Test
+    void finalPayloadConstructionErrorClearsTheStillOwnedBytes() throws Exception {
+        AssertionError fatal = new AssertionError("staged-construction-fatal");
+        AtomicReference<byte[]> attemptedBytes = new AtomicReference<>();
+        BoundedMigrationExportBuffer buffer = new BoundedMigrationExportBuffer(64,
+                (metadata, bytes, size) -> {
+                    attemptedBytes.set(bytes);
+                    throw fatal;
+                });
+        buffer.write("sensitive-payload".getBytes(StandardCharsets.UTF_8));
+
+        assertThatThrownBy(() -> buffer.finish(new ExportResponse("export.env", "text/plain")))
+                .isSameAs(fatal);
+
+        assertThat(attemptedBytes.get()).containsOnly((byte) 0);
+    }
+
+    @Test
+    void ownerCloseIsTerminalAndCannotRegrowOrFinalizeTheBuffer() throws Exception {
+        BoundedMigrationExportBuffer buffer = new BoundedMigrationExportBuffer(64);
+        buffer.write("sensitive-payload".getBytes(StandardCharsets.UTF_8));
+
+        buffer.close();
+        buffer.close();
+
+        assertThatThrownBy(() -> buffer.write(1)).isInstanceOf(IOException.class);
+        assertThatThrownBy(() -> buffer.finish(new ExportResponse("export.env", "text/plain")))
+                .isInstanceOf(IllegalStateException.class);
+    }
+}

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/setup/workflow/PreparedMigrationExportTest.java

@@ -0,0 +1,208 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+
+package org.apache.hertzbeat.manager.setup.workflow;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hertzbeat.manager.setup.api.SetupApiContract.ExportResponse;
+import org.apache.hertzbeat.manager.setup.config.SecretValue;
+import org.junit.jupiter.api.Test;
+
+/** Ownership, bounded staging, and one-shot lifecycle contracts for migration export. */
+class PreparedMigrationExportTest {
+
+    private static final ExportResponse METADATA =
+            new ExportResponse("hertzbeat-migration.env", "text/plain");
+
+    @Test
+    void stagesIntoOwnedBytesBeforeWritingAndClearsItsIndependentSecret() throws Exception {
+        SecretValue borrowed = SecretValue.of("export-secret");
+        AtomicReference<SecretValue> rendererSecret = new AtomicReference<>();
+        MigrationExportRenderer renderer = (secret, output) -> {
+            rendererSecret.set(secret);
+            output.write("rendered".getBytes(StandardCharsets.UTF_8));
+        };
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed, renderer);
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+        try (StagedMigrationExport staged = prepared.stage()) {
+            assertSame(METADATA, staged.metadata());
+            assertThat(staged.size()).isEqualTo(8);
+            assertThat(output.size()).isZero();
+            staged.writeTo(output);
+        }
+
+        assertThat(output.toString(StandardCharsets.UTF_8)).isEqualTo("rendered");
+        assertThat(rendererSecret.get()).isNotSameAs(borrowed);
+        assertThat(rendererSecret.get().copy()).containsOnly('\0');
+        assertThat(borrowed.copy()).containsExactly("export-secret".toCharArray());
+        prepared.close();
+        borrowed.close();
+    }
+
+    @Test
+    void rendererFailureProducesNoStagedPayloadAndClearsSecret() {
+        SecretValue borrowed = SecretValue.of("failure-secret");
+        AtomicReference<SecretValue> rendererSecret = new AtomicReference<>();
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> {
+                    rendererSecret.set(secret);
+                    output.write("partial-private".getBytes(StandardCharsets.UTF_8));
+                    throw new IOException("private-renderer-detail");
+                });
+
+        assertThatThrownBy(prepared::stage)
+                .isInstanceOf(IOException.class)
+                .hasMessageNotContaining("private-renderer-detail")
+                .hasNoCause();
+        assertThat(rendererSecret.get().copy()).containsOnly('\0');
+        assertThat(borrowed.copy()).containsExactly("failure-secret".toCharArray());
+        borrowed.close();
+    }
+
+    @Test
+    void rejectsPayloadPastTheFixedLimitWithoutReturningPartialBytes() {
+        SecretValue borrowed = SecretValue.of("bounded-secret");
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> output.write(new byte[PreparedMigrationExport.MAX_EXPORT_BYTES + 1]));
+
+        assertThatThrownBy(prepared::stage)
+                .isInstanceOf(IOException.class)
+                .hasNoCause()
+                .hasMessageNotContaining("bounded-secret");
+        borrowed.close();
+    }
+
+    @Test
+    void rendererCloseAtEndDoesNotDiscardThePreparedPayload() throws Exception {
+        SecretValue borrowed = SecretValue.of("close-at-end-secret");
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> {
+                    output.write("complete".getBytes(StandardCharsets.UTF_8));
+                    output.close();
+                });
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+        try (StagedMigrationExport staged = prepared.stage()) {
+            staged.writeTo(output);
+        }
+
+        assertThat(output.toString(StandardCharsets.UTF_8)).isEqualTo("complete");
+        borrowed.close();
+    }
+
+    @Test
+    void rendererCloseAndContinuedWriteRemainValid() throws Exception {
+        SecretValue borrowed = SecretValue.of("close-then-write-secret");
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> {
+                    output.close();
+                    output.write("after-close".getBytes(StandardCharsets.UTF_8));
+                });
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+        try (StagedMigrationExport staged = prepared.stage()) {
+            staged.writeTo(output);
+        }
+
+        assertThat(output.toString(StandardCharsets.UTF_8)).isEqualTo("after-close");
+        borrowed.close();
+    }
+
+    @Test
+    void closeBeforeStageClearsOwnedSecretAndRejectsLaterConsumption() {
+        SecretValue borrowed = SecretValue.of("unused-secret");
+        AtomicReference<SecretValue> rendererSecret = new AtomicReference<>();
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> rendererSecret.set(secret));
+
+        prepared.close();
+        prepared.close();
+
+        assertThatThrownBy(prepared::stage)
+                .isInstanceOf(IllegalStateException.class)
+                .hasMessageNotContaining("unused-secret");
+        assertThat(rendererSecret).hasValue(null);
+        borrowed.close();
+    }
+
+    @Test
+    void metadataAndDiagnosticsNeverExposeRendererOrSecret() {
+        SecretValue borrowed = SecretValue.of("diagnostic-secret");
+        MigrationExportRenderer renderer = (secret, output) -> { };
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed, renderer);
+
+        assertSame(METADATA, prepared.metadata());
+        assertThat(prepared.toString())
+                .doesNotContain("diagnostic-secret")
+                .doesNotContain(renderer.getClass().getName());
+        prepared.close();
+        borrowed.close();
+    }
+
+    @Test
+    void firstRendererErrorRemainsPrimaryWhileOwnedMaterialIsCleared() {
+        SecretValue borrowed = SecretValue.of("fatal-secret");
+        AtomicReference<SecretValue> rendererSecret = new AtomicReference<>();
+        AssertionError fatal = new AssertionError("renderer-fatal");
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> {
+                    rendererSecret.set(secret);
+                    throw fatal;
+                });
+
+        assertThatThrownBy(prepared::stage).isSameAs(fatal).hasNoCause();
+        assertThat(rendererSecret.get().copy()).containsOnly('\0');
+        borrowed.close();
+    }
+
+    @Test
+    void stagedBytesAreClearedOnCloseAndCannotBeWrittenAgain() throws Exception {
+        SecretValue borrowed = SecretValue.of("staged-secret");
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> output.write("payload".getBytes(StandardCharsets.UTF_8)));
+        StagedMigrationExport staged = prepared.stage();
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+        staged.close();
+        staged.close();
+
+        assertThatThrownBy(() -> staged.writeTo(output))
+                .isInstanceOf(IllegalStateException.class)
+                .hasNoCause();
+        assertThat(output.size()).isZero();
+        borrowed.close();
+    }
+
+    @Test
+    void bufferAllocationErrorStillClearsOwnedSecretAndEndsPreparation() throws Exception {
+        SecretValue borrowed = SecretValue.of("allocation-secret");
+        AssertionError fatal = new AssertionError("buffer-allocation-fatal");
+        PreparedMigrationExport prepared = PreparedMigrationExport.prepare(METADATA, borrowed,
+                (secret, output) -> { }, limit -> {
+                    throw fatal;
+                });
+        Field secretField = PreparedMigrationExport.class.getDeclaredField("ownedSecret");
+        secretField.setAccessible(true);
+        SecretValue ownedSecret = (SecretValue) secretField.get(prepared);
+
+        assertThatThrownBy(prepared::stage).isSameAs(fatal);
+
+        assertThat(ownedSecret.copy()).containsOnly('\0');
+        assertThat(prepared.toString()).contains("CLOSED").doesNotContain("allocation-secret");
+        prepared.close();
+        borrowed.close();
+    }
+}