feat(deployment): report terminal deployment status to source account after endpoint switch

After a successful endpoint switch (or FAILED_UNABLE_TO_ROLLBACK), the main
MQTT client is connected to the destination endpoint and cannot report status
back to the source endpoint's IoT Job or deployment shadow. This adds source
endpoint status reporting via a standalone MQTT connection.

- EndpointSwitchState: new class managing persisted state (source endpoint,
  deployment ID) for in-flight endpoint-switch deployments
- IotJobsHelper: consumer detects endpoint-switch deployment, publishes
  SUCCEEDED to source IoT Job via standalone MQTT, clears keys
- ShadowDeploymentListener: same pattern for shadow deployments (publishes
  unconditional shadow update without version field)
- StandaloneMqttConnector.publish(): reusable publish with exponential backoff
- IN_PROGRESS guard: dropped after nucleus restart to avoid stale publishes
  to destination where the IoT Job/shadow doesn't exist

Author: Sahith Reddy Adudodla

src/main/java/com/aws/greengrass/deployment/DeploymentConfigMerger.java

@@ -62,10 +62,7 @@ public class DeploymentConfigMerger {
     public static final String DEPLOYMENT_ID_LOG_KEY = "deploymentId";
     public static final String SERVICE_NAME_LOG_KEY = "serviceName";
     protected static final int WAIT_SVC_START_POLL_INTERVAL_MILLISEC = 1000;
-    static final long DEFAULT_PREFLIGHT_MQTT_TIMEOUT_MS = 60_000;
-    static final String STANDALONE_MQTT_TIMEOUT_KEY = "standaloneMqttTimeoutMs";
     private static final String ENDPOINT_LOG_KEY = "endpoint";
-    private static final String ENDPOINT_SWITCH_CLIENT_SUFFIX = "#endpoint-switch";
 
     private static final Logger logger = LogManager.getLogger(DeploymentConfigMerger.class);
 
@@ -161,10 +158,20 @@ private void updateActionForDeployment(Map<String, Object> newConfig, Deployment
 
         // Endpoint-switch deployments: pre-flight connectivity check, then persist source endpoint.
         // Pre-flight runs BEFORE persisting so that on failure, no device state has changed.
-        // Source endpoint is stored under services.DeploymentService.runtime (internal deployment state,
-        // not customer config). Persisted to config.tlog so it survives device crashes mid-endpoint-switch.
-        // Step 5 uses this value to report deployment status back to the source account.
-        // Cleared after status reporting; stale keys are harmless and overwritten by the next switch.
+        //
+        // Before switching the IoT endpoint, save two values to runtime config:
+        //
+        // 1. sourceIotDataEndpoint — the current IoT data endpoint (where the device is connected
+        //    now). After the switch, the status consumer uses this to publish the deployment result
+        //    back to the originating account via a standalone MQTT connection.
+        //
+        // 2. sourceDeploymentId — this deployment's ID (job ID, config ARN, or UUID depending on
+        //    type). The status consumer matches this against the persisted status entry to identify
+        //    which entry belongs to the endpoint-switch deployment.
+        //
+        // Both are stored under services.DeploymentService.runtime (internal deployment state, not
+        // customer config) and persisted to config.tlog (survives crashes). Removed by the status
+        // consumer after it successfully publishes the deployment result to the source endpoint.
         EndpointSwitchPreflightValidator preflightValidator =
                 kernel.getContext().get(EndpointSwitchPreflightValidator.class);
         if (isEndpointSwitchDeployment(nucleusConfig, deviceConfiguration)) {
@@ -180,24 +187,24 @@ private void updateActionForDeployment(Map<String, Object> newConfig, Deployment
                     .kv(ENDPOINT_LOG_KEY, newDataEndpoint)
                     .log("Starting pre-flight MQTT connectivity check");
             long preflightTimeout = Coerce.toLong(deviceConfiguration.getMQTTNamespace()
-                    .findOrDefault(DEFAULT_PREFLIGHT_MQTT_TIMEOUT_MS, STANDALONE_MQTT_TIMEOUT_KEY));
+                    .findOrDefault(EndpointSwitchState.DEFAULT_STANDALONE_MQTT_TIMEOUT_MS,
+                            EndpointSwitchState.STANDALONE_MQTT_TIMEOUT_KEY));
             if (preflightTimeout <= 0) {
                 logger.atWarn().kv("configured", preflightTimeout)
-                        .kv("using", DEFAULT_PREFLIGHT_MQTT_TIMEOUT_MS)
+                        .kv("using", EndpointSwitchState.DEFAULT_STANDALONE_MQTT_TIMEOUT_MS)
                         .log("Invalid standaloneMqttTimeoutMs, using default");
-                preflightTimeout = DEFAULT_PREFLIGHT_MQTT_TIMEOUT_MS;
+                preflightTimeout = EndpointSwitchState.DEFAULT_STANDALONE_MQTT_TIMEOUT_MS;
             }
             if (!preflightValidator.verifyMqttConnectivity(totallyCompleteFuture, newDataEndpoint,
-                    ENDPOINT_SWITCH_CLIENT_SUFFIX, preflightTimeout)) {
+                    EndpointSwitchState.ENDPOINT_SWITCH_CLIENT_SUFFIX, preflightTimeout)) {
                 return;
             }
             logger.atInfo().setEventType(MERGE_CONFIG_EVENT_KEY)
                     .kv(ENDPOINT_LOG_KEY, newDataEndpoint)
                     .log("Pre-flight MQTT connectivity check passed");
 
             // Persist source endpoint only after pre-flight succeeds — no state change on failure
-            kernel.getContext().get(DeploymentService.class).getRuntimeConfig()
-                    .lookup(DeploymentService.SOURCE_IOT_DATA_ENDPOINT_KEY).withValue(currentDataEndpoint);
+            kernel.getContext().get(EndpointSwitchState.class).persist(currentDataEndpoint, deployment.getId());
         }
 
         // Pre-flight credential endpoint check — uses the device certificate for mTLS authentication,

src/main/java/com/aws/greengrass/deployment/DeploymentService.java

@@ -98,7 +98,6 @@ public class DeploymentService extends GreengrassService {
     public static final String GROUP_TO_LAST_DEPLOYMENT_CONFIG_ARN_KEY = "configArn";
     public static final String GROUP_TO_ROOT_COMPONENTS_VERSION_KEY = "version";
     public static final String GROUP_TO_ROOT_COMPONENTS_GROUP_CONFIG_ARN = "groupConfigArn";
-    public static final String SOURCE_IOT_DATA_ENDPOINT_KEY = "sourceIotDataEndpoint";
     public static final String GROUP_TO_ROOT_COMPONENTS_GROUP_NAME = "groupConfigName";
     public static final String DEPLOYMENT_DETAILED_STATUS_KEY = "detailed-deployment-status";
     public static final String DEPLOYMENT_FAILURE_CAUSE_KEY = "deployment-failure-cause";
@@ -422,6 +421,7 @@ private void finishCurrentDeployment() throws InterruptedException {
             logger.atInfo().kv(DEPLOYMENT_ID_LOG_KEY_NAME, currentDeploymentTaskMetadata.getDeploymentId())
                     .kv(GG_DEPLOYMENT_ID_LOG_KEY_NAME, ggDeploymentId).log("Deployment task is cancelled");
         }
+
         // Setting this to null to indicate there is no current deployment being processed
         // Did not use optionals over null due to performance
         currentDeploymentTaskMetadata = null;
@@ -1059,6 +1059,7 @@ public Set<String> getAllGroupNames() {
         return allGroupNames;
     }
 
+
     /**
      * Checks whether a component is a root component or not.
      *

src/main/java/com/aws/greengrass/deployment/EndpointSwitchState.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.aws.greengrass.deployment;
+
+import com.aws.greengrass.config.Topic;
+import com.aws.greengrass.config.Topics;
+import com.aws.greengrass.util.Coerce;
+import com.aws.greengrass.util.Utils;
+
+import javax.inject.Inject;
+
+/**
+ * Manages persisted state for in-flight endpoint-switch deployments.
+ *
+ * <p>Before an endpoint switch, {@link DeploymentConfigMerger} persists the source IoT data endpoint and
+ * deployment ID into DeploymentService's runtime config. After the switch completes, the status consumers
+ * ({@link IotJobsHelper}, {@link ShadowDeploymentListener}) use this class to identify the endpoint-switch
+ * deployment and report terminal status back to the source endpoint via a standalone MQTT connection.</p>
+ *
+ * <p>Keys are stored under {@code services.DeploymentService.runtime} and persisted to config.tlog
+ * (survives crashes). Cleared after terminal status is reported.</p>
+ */
+public class EndpointSwitchState {
+
+    static final String SOURCE_IOT_DATA_ENDPOINT_KEY = "sourceIotDataEndpoint";
+    static final String SOURCE_DEPLOYMENT_ID_KEY = "sourceDeploymentId";
+    static final String ENDPOINT_SWITCH_CLIENT_SUFFIX = "#endpoint-switch";
+    static final long DEFAULT_STANDALONE_MQTT_TIMEOUT_MS = 60_000;
+    static final String STANDALONE_MQTT_TIMEOUT_KEY = "standaloneMqttTimeoutMs";
+
+    private final DeploymentService deploymentService;
+
+    @Inject
+    EndpointSwitchState(DeploymentService deploymentService) {
+        this.deploymentService = deploymentService;
+    }
+
+    /**
+     * Check if the given deployment is the in-flight endpoint-switch deployment by matching
+     * the deployment ID against the persisted source deployment ID.
+     *
+     * @param deploymentId the deployment ID to check (job ID for IoT Jobs, config ARN for Shadow)
+     * @return true if this is the endpoint-switch deployment
+     */
+    public boolean isEndpointSwitchDeployment(String deploymentId) {
+        String sourceEndpoint = getSourceIotDataEndpoint();
+        if (sourceEndpoint == null) {
+            return false;
+        }
+        String sourceDepId = getSourceDeploymentId();
+        return deploymentId != null && deploymentId.equals(sourceDepId);
+    }
+
+    /**
+     * Read the IoT data endpoint where the endpoint-switch deployment originated.
+     * This is the endpoint the device was connected to before the switch, used to report terminal
+     * deployment status back to the source endpoint via a standalone MQTT connection.
+     *
+     * @return the source IoT data endpoint, or null if no endpoint switch is in progress
+     */
+    public String getSourceIotDataEndpoint() {
+        String sourceEndpoint = Coerce.toString(getRuntimeConfig().find(SOURCE_IOT_DATA_ENDPOINT_KEY));
+        if (Utils.isEmpty(sourceEndpoint)) {
+            return null;
+        }
+        return sourceEndpoint;
+    }
+
+    /**
+     * Persist the source endpoint and deployment ID before applying the endpoint switch.
+     * Called by {@link DeploymentConfigMerger} after pre-flight passes.
+     *
+     * @param sourceEndpoint the current IoT data endpoint (before switch)
+     * @param deploymentId   the deployment ID initiating the switch
+     */
+    public void persist(String sourceEndpoint, String deploymentId) {
+        getRuntimeConfig().lookup(SOURCE_IOT_DATA_ENDPOINT_KEY).withValue(sourceEndpoint);
+        getRuntimeConfig().lookup(SOURCE_DEPLOYMENT_ID_KEY).withValue(deploymentId);
+    }
+
+    /**
+     * Remove the persisted source IoT data endpoint and deployment ID from runtime config.
+     * Called after the endpoint-switch deployment's terminal status has been published to the
+     * source endpoint (or publish failed — keys are cleared regardless to avoid stale state).
+     */
+    public void clear() {
+        Topic sourceEndpointTopic = getRuntimeConfig().find(SOURCE_IOT_DATA_ENDPOINT_KEY);
+        if (sourceEndpointTopic != null) {
+            sourceEndpointTopic.remove();
+        }
+        Topic sourceDeploymentIdTopic = getRuntimeConfig().find(SOURCE_DEPLOYMENT_ID_KEY);
+        if (sourceDeploymentIdTopic != null) {
+            sourceDeploymentIdTopic.remove();
+        }
+    }
+
+    private String getSourceDeploymentId() {
+        return Coerce.toString(getRuntimeConfig().find(SOURCE_DEPLOYMENT_ID_KEY));
+    }
+
+    private Topics getRuntimeConfig() {
+        return deploymentService.getRuntimeConfig();
+    }
+}

src/main/java/com/aws/greengrass/deployment/IotJobsClientWrapper.java

@@ -40,7 +40,7 @@
 @SuppressWarnings("PMD.AvoidCatchingGenericException")
 @SuppressFBWarnings("NM_METHOD_NAMING_CONVENTION")
 public class IotJobsClientWrapper extends IotJobsClient {
-    private static final String UPDATE_JOB_TOPIC =
+    static final String UPDATE_JOB_TOPIC =
             "$aws/things/%s/jobs/%s/namespace-aws-gg-deployment/update";
     static final String JOB_UPDATE_ACCEPTED_TOPIC =
             "$aws/things/%s/jobs/%s/namespace-aws-gg-deployment/update/accepted";

src/main/java/com/aws/greengrass/deployment/IotJobsHelper.java

@@ -17,7 +17,9 @@
 import com.aws.greengrass.logging.api.Logger;
 import com.aws.greengrass.logging.impl.LogManager;
 import com.aws.greengrass.mqttclient.MqttClient;
+import com.aws.greengrass.mqttclient.StandaloneMqttConnector;
 import com.aws.greengrass.mqttclient.WrapperMqttClientConnection;
+import com.aws.greengrass.security.SecurityService;
 import com.aws.greengrass.status.FleetStatusService;
 import com.aws.greengrass.status.model.Trigger;
 import com.aws.greengrass.util.Coerce;
@@ -26,6 +28,8 @@
 import com.aws.greengrass.util.SerializerFactory;
 import com.aws.greengrass.util.Utils;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.gson.GsonBuilder;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
@@ -46,6 +50,7 @@
 import software.amazon.awssdk.iot.iotjobs.model.UpdateJobExecutionRequest;
 import software.amazon.awssdk.iot.iotjobs.model.UpdateJobExecutionSubscriptionRequest;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.HashMap;
@@ -97,6 +102,10 @@ public class IotJobsHelper implements InjectionActions {
             "Interrupted while subscribing to Iot jobs event notifications topic";
     private static final String JOB_ID_LOG_KEY_NAME = "JobId";
     private static final String NEXT_JOB_LITERAL = "$next";
+    /**
+     * IoT Jobs status detail values are limited to 1024 characters by the service.
+     */
+    private static final int JOB_STATUS_DETAIL_MAX_LENGTH = 1024;
     // Sometimes when we are notified that a new job is queued and request the next pending job document immediately,
     // we get an empty response. This unprocessedJobs is to track the number of new queued jobs that we are notified
     // with, and keep retrying the request until we get a non-empty response.
@@ -361,6 +370,34 @@ private void setupCommWithIotJobs(Boolean isConfigurationUpdate) {
     private Boolean deploymentStatusChanged(Map<String, Object> deploymentDetails) {
         String jobId = (String) deploymentDetails.get(DEPLOYMENT_ID_KEY_NAME);
         String status = (String) deploymentDetails.get(DEPLOYMENT_STATUS_KEY_NAME);
+
+        // Endpoint-switch deployment: report SUCCEEDED to the source endpoint via standalone MQTT.
+        //
+        // After a successful switch the main MQTT client is on the destination endpoint, so the
+        // IoT Job at the source endpoint can only be updated via a standalone connection.
+        //
+        // FAILED (rollback): the device is still on the source endpoint (sourceEndpoint ==
+        // currentEndpoint), so the block clears keys and falls through — the existing
+        // updateJobStatus() below publishes FAILED via the main MQTT client which is still
+        // connected to the source endpoint.
+        //
+        // IN_PROGRESS: only reaches this block after a nucleus restart (the source endpoint
+        // already received IN_PROGRESS before the restart). Dropped to avoid publishing to the
+        // destination endpoint where the IoT Job doesn't exist.
+        EndpointSwitchState endpointSwitchState = kernel.getContext().get(EndpointSwitchState.class);
+        if (endpointSwitchState.isEndpointSwitchDeployment(jobId)) {
+            if (JobStatus.IN_PROGRESS.toString().equals(status)) {
+                return true;
+            }
+            String sourceEndpoint = endpointSwitchState.getSourceIotDataEndpoint();
+            if (!sourceEndpoint.equals(Coerce.toString(deviceConfiguration.getIotDataEndpoint()))) {
+                publishStatusToSourceEndpoint(sourceEndpoint, jobId, deploymentDetails);
+                endpointSwitchState.clear();
+                return true;
+            }
+            endpointSwitchState.clear();
+        }
+
         Map<String, Object> statusDetails =
                 (Map<String, Object>) deploymentDetails.get(DEPLOYMENT_STATUS_DETAILS_KEY_NAME);
         HashMap<String, String> jobStatusDetails = new HashMap<>();
@@ -435,7 +472,8 @@ public void updateJobStatus(String jobId, JobStatus status, HashMap<String, Stri
                 });
 
         // Truncate status detail map values longer than the 1024 characters limit
-        statusDetailsMap.entrySet().forEach(e -> statusDetailsMap.put(e.getKey(), Utils.truncate(e.getValue(), 1024)));
+        statusDetailsMap.entrySet().forEach(e -> statusDetailsMap.put(e.getKey(),
+                Utils.truncate(e.getValue(), JOB_STATUS_DETAIL_MAX_LENGTH)));
 
         UpdateJobExecutionRequest updateJobRequest = new UpdateJobExecutionRequest();
         updateJobRequest.jobId = jobId;
@@ -664,6 +702,61 @@ private void evaluateCancellationAndCancelDeploymentIfNeeded() {
         }
     }
 
+    @SuppressWarnings("PMD.AvoidCatchingGenericException")
+    @SuppressFBWarnings("REC_CATCH_EXCEPTION")
+    private void publishStatusToSourceEndpoint(String sourceEndpoint, String jobId,
+                                                  Map<String, Object> deploymentDetails) {
+        String status = (String) deploymentDetails.get(DEPLOYMENT_STATUS_KEY_NAME);
+        Map<String, Object> statusDetails =
+                (Map<String, Object>) deploymentDetails.get(DEPLOYMENT_STATUS_DETAILS_KEY_NAME);
+        HashMap<String, String> jobStatusDetails = new HashMap<>();
+        statusDetails.forEach((k, v) -> {
+            if (v instanceof String) {
+                jobStatusDetails.put(k, (String) v);
+            } else {
+                try {
+                    jobStatusDetails.put(k, SerializerFactory.getFailSafeJsonObjectMapper().writeValueAsString(v));
+                } catch (JsonProcessingException e) {
+                    logger.atWarn().setCause(e).log("Failed to serialize status detail");
+                }
+            }
+        });
+
+        jobStatusDetails.entrySet().forEach(e -> jobStatusDetails.put(e.getKey(),
+                Utils.truncate(e.getValue(), JOB_STATUS_DETAIL_MAX_LENGTH)));
+
+        String thing = Coerce.toString(deviceConfiguration.getThingName());
+        String topic = String.format(IotJobsClientWrapper.UPDATE_JOB_TOPIC, thing, jobId);
+        long connectTimeout = Coerce.toLong(deviceConfiguration.getMQTTNamespace()
+                .findOrDefault(EndpointSwitchState.DEFAULT_STANDALONE_MQTT_TIMEOUT_MS,
+                        EndpointSwitchState.STANDALONE_MQTT_TIMEOUT_KEY));
+        if (connectTimeout <= 0) {
+            connectTimeout = EndpointSwitchState.DEFAULT_STANDALONE_MQTT_TIMEOUT_MS;
+        }
+        try (StandaloneMqttConnector connector = StandaloneMqttConnector.of(
+                kernel.getContext().get(SecurityService.class), deviceConfiguration,
+                sourceEndpoint, EndpointSwitchState.ENDPOINT_SWITCH_CLIENT_SUFFIX)) {
+            connector.connect(connectTimeout);
+            UpdateJobExecutionRequest request = new UpdateJobExecutionRequest();
+            request.status = JobStatus.valueOf(status);
+            request.statusDetails = jobStatusDetails;
+            request.thingName = thing;
+            request.jobId = jobId;
+            // Use Gson with HTML escaping disabled to match the IoT Jobs SDK's wire format
+            // (IotJobsClientWrapper uses the same GsonBuilder configuration)
+            byte[] payload = new GsonBuilder().disableHtmlEscaping().create()
+                    .toJson(request).getBytes(StandardCharsets.UTF_8);
+            connector.publish(topic, payload, QualityOfService.AT_LEAST_ONCE,
+                    EndpointSwitchState.DEFAULT_STANDALONE_MQTT_TIMEOUT_MS);
+            logger.atInfo().kv("jobId", jobId).kv("status", status)
+                    .kv("sourceEndpoint", sourceEndpoint)
+                    .log("Reported terminal job status to source endpoint");
+        } catch (Exception e) {
+            logger.atWarn().kv("jobId", jobId).kv("status", status).kv("sourceEndpoint", sourceEndpoint)
+                    .setCause(e).log("Failed to report terminal job status to source endpoint");
+        }
+    }
+
     public static class WrapperMqttConnectionFactory {
         public WrapperMqttClientConnection getAwsIotMqttConnection(MqttClient mqttClient) {
             return new WrapperMqttClientConnection(mqttClient);