core: stop worker when unable to process requests after infra load

The worker is able to process requests if infra load:
* ... went fine, happy path (just process request)
* ... failed with a perennial error (expected behavior is reject requests
  on the same version of the same infra)

If infra load failed with a temporary error (and after possible retries),
better just stop and let orchestrator decide to retry or not.

No retry implemented so far.

Signed-off-by: Pierre-Etienne Bougué <[email protected]>

Author: bougue-pe

core/osrd-reporting/src/main/java/fr/sncf/osrd/reporting/exceptions/ErrorType.java

@@ -26,11 +26,12 @@ public enum ErrorType {
     StrictWarningError("strict_warning", "Warning was reported with strict mode enabled", ErrorCause.USER),
     AssertionError("assert_error", "assert check failed", ErrorCause.INTERNAL),
     UnknownError("unknown_error", "An unknown exception was thrown", ErrorCause.INTERNAL),
-    InfraSoftLoadingError("infra_loading:soft_error", "soft error while loading new infra", ErrorCause.USER),
+    InfraSoftLoadingError("infra_loading:soft_error", "soft error while loading new infra", ErrorCause.USER, false),
     InfraHardLoadingError("infra_loading:hard_error", "hard error while loading new infra", ErrorCause.USER),
     InfraHardError("infra:hard_error", "hard error while parsing infra", ErrorCause.USER),
     InfraLoadingCacheException("infra_loading:cache_exception", "cached exception", ErrorCause.INTERNAL),
-    InfraLoadingInvalidStatusException("infra_loading:invalid_status", "Status doesn’t exist", ErrorCause.INTERNAL),
+    InfraLoadingInvalidStatusException(
+            "infra_loading:invalid_status", "infra not loaded correctly", ErrorCause.INTERNAL),
     InfraInvalidStatusWhileWaitingStable(
             "infra_loading:invalid_status_waiting_stable", "invalid status after waitUntilStable", ErrorCause.INTERNAL),
     InfraInvalidVersionException("infra:invalid_version", "Invalid infra version", ErrorCause.USER),

core/src/main/java/fr/sncf/osrd/api/ExceptionHandler.java

@@ -1,6 +1,7 @@
 package fr.sncf.osrd.api;
 
 import fr.sncf.osrd.reporting.exceptions.ErrorCause;
+import fr.sncf.osrd.reporting.exceptions.ErrorType;
 import fr.sncf.osrd.reporting.exceptions.OSRDError;
 import org.takes.Response;
 import org.takes.rs.RsJson;
@@ -9,11 +10,16 @@
 
 public class ExceptionHandler {
 
-    /** Handles an exception, returns an HTTP response with all relevant information */
-    public static Response handle(Throwable ex) {
+    /** Handles an exception, returns an HTTP response with all relevant information or
+     * re-throw if exception is an OSRDError of type InfraSoftLoadingError */
+    public static Response handle(Throwable ex) throws OSRDError {
         ex.printStackTrace();
-        if (ex instanceof OSRDError) return toResponse((OSRDError) ex);
-        else if (ex instanceof AssertionError) return toResponse(OSRDError.newAssertionWrapper((AssertionError) ex));
+        if (ex instanceof OSRDError osrdError) {
+            if (osrdError.osrdErrorType == ErrorType.InfraSoftLoadingError) {
+                throw osrdError;
+            }
+            return toResponse(osrdError);
+        } else if (ex instanceof AssertionError) return toResponse(OSRDError.newAssertionWrapper((AssertionError) ex));
         else {
             return toResponse(OSRDError.newUnknownError(ex));
         }

core/src/main/java/fr/sncf/osrd/api/InfraManager.java

@@ -111,13 +111,22 @@ private FullInfra downloadInfra(InfraCacheEntry cacheEntry, String infraId, Diag
             RJSInfra rjsInfra;
             String version;
             try (var response = httpClient.newCall(request).execute()) {
-                if (!response.isSuccessful()) throw new UnexpectedHttpResponse(response);
+                if (!response.isSuccessful()) {
+                    if (response.code() != 404) {
+                        throw new UnexpectedHttpResponse(response);
+                    } else {
+                        logger.info("Infra not found (deleted) on supplier middleware");
+                        throw OSRDError.newInfraLoadingError(
+                                ErrorType.InfraHardLoadingError, "Infra not found (deleted) on supplier middleware");
+                    }
+                }
 
                 // Parse the response
                 logger.info("parsing the JSON of {}", request.url());
                 cacheEntry.transitionTo(InfraStatus.PARSING_JSON);
                 version = response.header("x-infra-version");
                 assert version != null : "missing x-infra-version header in railjson response";
+                cacheEntry.version = version;
                 rjsInfra = RJSInfra.adapter.fromJson(response.body().source());
             }
 
@@ -137,11 +146,11 @@ private FullInfra downloadInfra(InfraCacheEntry cacheEntry, String infraId, Diag
             // Cache the infra
             logger.info("successfully cached {}", request.url());
             cacheEntry.infra = new FullInfra(rawInfra, loadedSignalInfra, blockInfra, signalingSimulator);
-            cacheEntry.version = version;
             cacheEntry.transitionTo(InfraStatus.CACHED);
             return cacheEntry.infra;
         } catch (IOException | UnexpectedHttpResponse | VirtualMachineError e) {
             cacheEntry.transitionTo(InfraStatus.TRANSIENT_ERROR, e);
+            // TODO: retry with an exponential backoff and jitter (use a concurrent Thread.sleep)
             throw OSRDError.newInfraLoadingError(ErrorType.InfraSoftLoadingError, cacheEntry.lastStatus.name(), e);
         } catch (Throwable e) {
             cacheEntry.transitionTo(InfraStatus.ERROR, e);

core/src/main/java/fr/sncf/osrd/cli/WorkerCommand.kt

@@ -12,6 +12,8 @@ import fr.sncf.osrd.api.api_v2.standalone_sim.SimulationEndpoint
 import fr.sncf.osrd.api.api_v2.stdcm.STDCMEndpointV2
 import fr.sncf.osrd.api.pathfinding.PathfindingBlocksEndpoint
 import fr.sncf.osrd.api.stdcm.STDCMEndpoint
+import fr.sncf.osrd.reporting.exceptions.ErrorType
+import fr.sncf.osrd.reporting.exceptions.OSRDError
 import fr.sncf.osrd.reporting.warnings.DiagnosticRecorderImpl
 import io.opentelemetry.api.GlobalOpenTelemetry
 import io.opentelemetry.context.Context
@@ -134,7 +136,26 @@ class WorkerCommand : CliCommand {
             connection.createChannel().use { channel -> reportActivity(channel, "started") }
 
             if (!ALL_INFRA) {
-                infraManager.load(infraId, null, diagnosticRecorder)
+                try {
+                    infraManager.load(infraId, null, diagnosticRecorder)
+                } catch (e: OSRDError) {
+                    if (e.osrdErrorType == ErrorType.InfraHardLoadingError) {
+                        logger.warn("Failed to load infra $infraId with a perennial error: $e")
+                        // go on and future requests will be consumed and rejected
+                    } else if (e.osrdErrorType == ErrorType.InfraSoftLoadingError) {
+                        logger.error("Failed to load infra $infraId with a temporary error: $e")
+                        // Stop worker and let another worker spawn eventually
+                        throw e
+                    } else {
+                        logger.error(
+                            "Failed to load infra $infraId with an unexpected OSRD Error: $e"
+                        )
+                        throw e
+                    }
+                } catch (t: Throwable) {
+                    logger.error("Failed to load infra $infraId with an unexpected exception: $t")
+                    throw t
+                }
             }
 
             connection.createChannel().use { channel -> reportActivity(channel, "ready") }
@@ -233,6 +254,10 @@ class WorkerCommand : CliCommand {
                             "ERROR, exception received"
                                 .toByteArray() // TODO: have a valid payload for uncaught exceptions
                         status = "core_error".encodeToByteArray()
+                        // Stop worker and let another worker spawn eventually
+                        if (t is OSRDError && t.osrdErrorType == ErrorType.InfraSoftLoadingError) {
+                            throw t
+                        }
                     } finally {
                         span.end()
                     }

core/src/main/kotlin/fr/sncf/osrd/api/api_v2/pathfinding/PathfindingBlocksEndpointV2.kt

@@ -61,15 +61,12 @@ class PathfindingBlocksEndpointV2(private val infraManager: InfraManager) : Take
         } catch (error: NoPathFoundException) {
             pathfindingLogger.info("No path found")
             RsJson(RsWithBody(pathfindingResponseAdapter.toJson(error.response)))
-        } catch (error: OSRDError) {
-            if (!error.osrdErrorType.isCacheable) {
-                ExceptionHandler.handle(error)
-            } else {
-                pathfindingLogger.info("Pathfinding failed: ${error.message}")
-                val response = PathfindingFailed(error)
+        } catch (ex: Throwable) {
+            if (ex is OSRDError && ex.osrdErrorType.isCacheable) {
+                pathfindingLogger.info("Pathfinding failed: ${ex.message}")
+                val response = PathfindingFailed(ex)
                 RsJson(RsWithBody(pathfindingResponseAdapter.toJson(response)))
             }
-        } catch (ex: Throwable) {
             ExceptionHandler.handle(ex)
         }
     }

core/src/main/kotlin/fr/sncf/osrd/api/api_v2/standalone_sim/SimulationEndpoint.kt

@@ -72,14 +72,11 @@ class SimulationEndpoint(
                     request.path.pathItemPositions,
                 )
             return RsJson(RsWithBody(simulationResponseAdapter.toJson(res)))
-        } catch (error: OSRDError) {
-            if (!error.osrdErrorType.isCacheable) {
-                return ExceptionHandler.handle(error)
-            } else {
-                val response = SimulationFailed(error)
+        } catch (ex: Throwable) {
+            if (ex is OSRDError && ex.osrdErrorType.isCacheable) {
+                val response = SimulationFailed(ex)
                 return RsJson(RsWithBody(simulationResponseAdapter.toJson(response)))
             }
-        } catch (ex: Throwable) {
             return ExceptionHandler.handle(ex)
         }
     }