diff --git a/docs/otel/message-telemetry-bridge.md b/docs/otel/message-telemetry-bridge.md new file mode 100644 index 0000000000..d7c68e8482 --- /dev/null +++ b/docs/otel/message-telemetry-bridge.md @@ -0,0 +1,92 @@ +# Optional message telemetry bridge + +The engine exposes one optional `MessageTelemetry.Provider`. It owns no SDK, telemetry registry, +queue carrier, exporter, configuration storage, or background thread. A plugin can use standard +OpenTelemetry Context/Scope and propagators behind this neutral Java contract. + +`SOURCE` surrounds `Channel.process`, including preprocessing, source transformation, destination +chains and synchronous postprocessing. `TRANSFORM`, `SEND`, and `RESPONSE` surround the existing +shared filter/transformer, validated send, and response-transformer execution methods. The original +method bodies, transaction boundaries and queue operations are retained. A source span's final +source-message status is distinct from a destination's final status. Send/response observations +also receive the actual returned response status before the caller applies queue status rules. + +`DESTINATION` surrounds each RECEIVED/PENDING connector in a destination chain and each non-null +acquired or held queue attempt. Completed SENT traversal creates no destination observation. +Start runs inside the existing DAO/queue cleanup boundary; finish runs after those cleanup +attempts, including the fallback status-lock release if DAO or queue cleanup throws. Queue retry waiting and optional queued +transformation are included. Immediate send retries stay inside their current destination scope; +each later queue attempt creates a fresh scope. Internally caught pre-send, cleanup and interrupt +failures are reported even if the message remains QUEUED. Existing DAO/queue failure behavior is +retained; the hook does not repair an existing engine cleanup failure or invent successful sends. + +The provider receives the existing connector message, including its maps. It must not change +message processing. The intended plugin-owned exceptions are private content-free ingress propagation data and +publication of documented scalar propagation values in the channel map; that plugin behavior is not implemented by this bridge. + +`beforeStore(message, sourceMap)` is a resource-free ingress callback immediately before the +source map is wrapped as read-only and the source connector/maps are first persisted. It may add +private content-free propagation data while preserving all application entries. It must not open +spans/scopes or require a later completion callback. The source message has its final identity, +but its source map has not yet been assigned; use the explicit map argument. Message-id allocation, +initial message insert/overwrite and failures before this point remain outside this observation. +The callback does not create a propagation format, cache, queue claim or restart policy in the +engine; those remain plugin responsibilities. Late map changes are not guaranteed to persist: +RAW storage skips later source-map updates while retaining initial raw durability. A plugin must account for that mode explicitly. + +Capture occurs before submission to the destination-chain and JavaScript executors. The provider +returns a resource-free context activator, not a replacement business task. The engine activates +on the worker and invokes the original callable exactly once, restoring context in a lexical +finally. Capture does not start a span or open a scope. Rejection and cancellation before start +therefore require no telemetry cleanup. Cancelled running work closes its scope on its own thread. + +Ordinary callback failures, including linkage/assertion errors, are isolated and yield one fixed +warning per installation, without exception/message content. A failed start/activation must clean +up its own partial work before throwing; the engine cannot recover resources a provider never +returned. VM errors and ThreadDeath propagate. An original engine fatal remains primary if its +failure-notification callback also throws a fatal; the callback fatal is suppressed when possible. +Java try-with-resources governs suppression at the existing detailed boundaries. Destination finish +also handles deliberately caught engine failures: the first observed engine fatal remains primary +for telemetry even when later DAO cleanup throws. Existing engine finally/catch behavior is preserved; +the first callback fatal wins a later callback fatal, and identical errors cannot self-suppress. +All observation cleanup occurs after the engine's original cleanup attempts. Closing a registration is idempotent +and cannot detach a newer installation. Already captured context activates through its original +provider; new stage observations inside that work select the current installation. The token +neither drains nor shuts down provider resources. + +## Behavior matrix and evidence + +| Case | Required behavior | Evidence / remaining scope | +| --- | --- | --- | +| No provider | Same no-op observation and original callable; no per-message bridge allocation | `MessageTelemetryTest` identity checks plus direct fast-path inspection | +| Duplicate install / repeated close / stale token | Reject overlap; old token never detaches replacement | Registration regression | +| Ordinary start, capture, activation, status, failure, close errors | Business task/result/exception preserved; cleanup attempted | Bridge fault matrix and real-channel callback-failure fixture | +| Fatal callbacks | VM error / ThreadDeath identity retained; original engine fatal wins a second failure-callback fatal; standard suppression on cleanup | Full six-surface fatal matrix and actual transformer regression | +| Initial source-map preparation | Same ingress thread, before first map persistence; source keys then become read-only; ordinary callback failure preserves completion | Actual queued-source barrier/readback with both later-map storage and initial-raw-only storage; final SENT, plus default/registration/fatal controls | +| Synchronous processing | Balanced source/destination/transform/send/response scopes; unchanged stored outcomes | Private Derby channel fixture | +| Parallel destinations | Source context crosses worker submission; destination maps are separate | Two real destination chains, one worker and one inline | +| Filter or source transformation error | Correct durable FILTERED/ERROR; no destination scopes | Private Derby negative paths | +| Preprocessor/destination filter, transform, validator or response errors | Original handled failure and correct source/destination stored status | Actual channel negatives; checked response/transform failure identity | +| Synchronous and queued destination retries | One scope per actual send; raw response and final status distinguished | Actual two-attempt retry cases with durable final SENT | +| Queued transformation / held retry interruption | Detailed work belongs to the acquired destination; interruption records failure without inventing a second send | Actual queue-thread and stored QUEUED controls | +| Completed SENT traversal / persisted PENDING | No scope for traversal; PENDING runs only destination/response scopes | Actual chain invocation and fresh Derby-loaded PENDING object | +| Queue start fatal after acquisition | Existing queue cleanup permits real retry; no send for the failed start | Actual first-start ThreadDeath, subsequent single send and durable SENT | +| Queue close fatal | Entry and status lock released before telemetry closes | Actual queue ownership/lock assertions and durable SENT | +| DAO close failure with a held queue status lock | Fallback unlock precedes telemetry completion; existing queue disposition failure remains unchanged | Real DAO close followed by an injected failure; lock state asserted outside the callback | +| Body fatal plus later DAO close and telemetry close failures | Coarse observation retains the first fatal; original engine catch/finally result retained | Actual queue and direct destination-chain composition, including a second fatal from DAO close | +| Destination failed/close fatal combinations | First fatal identity retained; close attempted once, including identical error object | Dedicated finish-composition controls | +| Source queue | Source scope starts on actual queue worker; normal completion | Private Derby queued-source fixture; parent continuity is a separate adapter concern | +| Script worker success / failure / interrupt | Actual JavaScript executor transfers and restores context; original exception/cancellation semantics | Executor tests plus real Rhino execution, exception and infinite-loop cancellation | +| Rejected / cancelled before start | No task execution or open telemetry scope | Controlled executor regressions | +| Cancelled after start | Worker restores its own prior context | Bridge and JavaScript executor interruption tests | +| Detach with captured task | Previously captured immutable context remains usable; new installation independent | Bridge detach/worker tests | +| Fatal ingress preparation | Existing dispatch exception wrapping, rollback and process-lock cleanup; next message still completes | Actual source callback fatal before first connector insertion | +| Partial provider start | Provider must restore any partial attachment before throwing | Explicit provider responsibility; adapter requires its own fault tests | +| Channel-map propagation, incoming HTTP, unsampled parents | Standard W3C extraction/injection; runtime context independent of editable maps | Initial plugin slice proved; new destination/carrier integration pending | +| Destination attempt parent; retries/refill/restart/nested channel/batch | Real queue/transaction behavior unchanged; documented parent policy | Further reduced-design integration pending; no durable carrier in this bridge | +| Actual instrumented HTTP/JDBC | Dependency spans share the channel context without duplicates | Initial slice proved official library interoperability; current destination integration and deployed agent acceptance remain separate | +| UI action-time config / retries / ambiguous writes | Explicit plugin-owned persistence and ownership guarantees | Configuration adaptation pending; old plugin cannot yet start on this engine | + +This is a reduced engine development slice, not a compatible plugin release or complete OTel acceptance. +Performance must be measured on the final integrated reduced implementation; previous benchmarks +of the large lifecycle SPI do not establish this bridge's overhead. diff --git a/donkey/build.gradle b/donkey/build.gradle index dfe4615c42..f243aafb4f 100644 --- a/donkey/build.gradle +++ b/donkey/build.gradle @@ -4,6 +4,7 @@ dependencies { implementation libs.bundles.donkey.main testImplementation libs.bundles.donkey.test + testRuntimeOnly files('conf', 'donkeydbconf') } def donkeyModelJar = tasks.register('donkeyModelJar', Jar) { diff --git a/donkey/src/main/java/com/mirth/connect/donkey/model/message/ConnectorMessage.java b/donkey/src/main/java/com/mirth/connect/donkey/model/message/ConnectorMessage.java index b253f1fafe..39f77c7547 100644 --- a/donkey/src/main/java/com/mirth/connect/donkey/model/message/ConnectorMessage.java +++ b/donkey/src/main/java/com/mirth/connect/donkey/model/message/ConnectorMessage.java @@ -10,6 +10,8 @@ package com.mirth.connect.donkey.model.message; import java.io.Serializable; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; import java.util.Calendar; import java.util.HashMap; import java.util.Map; @@ -19,6 +21,48 @@ @XStreamAlias("connectorMessage") public class ConnectorMessage implements Serializable { + // Preserve the serial form of the pre-dispatch-prototype native class. + private static final long serialVersionUID = 8556587410415698618L; + + /** Optional content-free scalar state. Neither key nor value may retain an SDK/provider. */ + private transient volatile TelemetrySlot telemetrySlot; + private static final VarHandle TELEMETRY_SLOT; + static { + try { + TELEMETRY_SLOT = MethodHandles.lookup().findVarHandle(ConnectorMessage.class, "telemetrySlot", TelemetrySlot.class); + } catch (ReflectiveOperationException e) { + throw new ExceptionInInitializerError(e); + } + } + + private static final class TelemetrySlot { + final Object owner; + final Object value; + TelemetrySlot(Object owner, Object value) { this.owner = owner; this.value = value; } + } + + /** Reserve before a callback; an older/reentrant callback cannot publish over a newer one. */ + public final Object reserveTelemetryContext(Object owner) { + if (owner == null) throw new IllegalArgumentException("telemetry_context_owner"); + TelemetrySlot selected = new TelemetrySlot(owner, null); + telemetrySlot = selected; + return selected; + } + + /** Null retires only this reservation. The opaque reservation is not the owner key. */ + public final boolean completeTelemetryContext(Object reservation, Object value) { + if (!(reservation instanceof TelemetrySlot)) return false; + TelemetrySlot selected = (TelemetrySlot) reservation; + TelemetrySlot next = value == null ? null : new TelemetrySlot(selected.owner, value); + return TELEMETRY_SLOT.compareAndSet(this, selected, next); + } + + /** Possession of an unrelated key cannot read or manufacture an installed provider's slot. */ + public final Object getTelemetryContext(Object owner) { + TelemetrySlot selected = telemetrySlot; + return selected != null && owner != null && selected.owner == owner ? selected.value : null; + } + private long messageId; private int metaDataId; private String channelId; diff --git a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/Channel.java b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/Channel.java index a78f2371f7..6653f54a1e 100644 --- a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/Channel.java +++ b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/Channel.java @@ -1492,6 +1492,8 @@ private ConnectorMessage createAndStoreSourceMessage(DonkeyDao dao, RawMessage r // Add the destination set to the source map sourceMap.put(Constants.DESTINATION_SET_KEY, destinationSet); + MessageTelemetry.beforeStore(sourceMessage, sourceMap); + // The source map is read-only so we wrap it in an unmodifiable map sourceMessage.setSourceMap(Collections.unmodifiableMap(sourceMap)); @@ -1613,6 +1615,13 @@ protected void queue(ConnectorMessage sourceMessage) { * @throws InterruptedException */ protected Message process(ConnectorMessage sourceMessage, boolean markAsProcessed) throws InterruptedException { + try (var observation = MessageTelemetry.start(MessageTelemetry.Stage.SOURCE, sourceMessage)) { + try { return processMessage(sourceMessage, markAsProcessed); } + catch (InterruptedException | RuntimeException | Error failure) { observation.failed(failure); throw failure; } + } + } + + private Message processMessage(ConnectorMessage sourceMessage, boolean markAsProcessed) throws InterruptedException { ThreadUtils.checkInterruptedStatus(); long messageId = sourceMessage.getMessageId(); @@ -1799,6 +1808,7 @@ protected Message process(ConnectorMessage sourceMessage, boolean markAsProcesse message.setChannelMap(new HashMap(sourceMessage.getChannelMap())); message.setResponseMap(new HashMap(sourceMessage.getResponseMap())); message.setRaw(raw); + MessageTelemetry.copyDispatchContext(sourceMessage, message); // store the new message, but we don't need to store the content because we will reference the source's encoded content dao.insertConnectorMessage(message, storageSettings.isStoreMaps(), true); @@ -1834,7 +1844,7 @@ protected Message process(ConnectorMessage sourceMessage, boolean markAsProcesse try { DestinationChain chain = enabledChains.get(i); chain.setName("Destination Chain Thread " + (i + 1) + " on " + name + " (" + channelId + ")"); - destinationChainTasks.add(channelExecutor.submit(chain)); + destinationChainTasks.add(channelExecutor.submit(MessageTelemetry.wrap(chain))); } catch (RejectedExecutionException e) { Thread.currentThread().interrupt(); throw new InterruptedException(); diff --git a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationChain.java b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationChain.java index 3a09743f36..4b28954c7d 100644 --- a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationChain.java +++ b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationChain.java @@ -100,106 +100,126 @@ private List doCall() throws InterruptedException { */ DonkeyDao dao = chainProvider.getDaoFactory().getDao(); + MessageTelemetry.Observation observation = null; + Throwable observationFailure = null; try { - Status previousStatus = message.getStatus(); - try { - switch (message.getStatus()) { - case RECEIVED: - /* - * Only transform the message if we're going to be dispatching it in the - * main processing thread, or if the queue thread will not be handling - * transformation - */ - if (destinationConnector.willAttemptSend() || !destinationConnector.includeFilterTransformerInQueue()) { - destinationConnector.transform(dao, message, previousStatus, true); - - // If the message status is QUEUED, send it to the destination connector - if (message.getStatus() == Status.QUEUED) { - String originalThreadName = Thread.currentThread().getName(); - try { - Thread.currentThread().setName(destinationConnector.getConnectorProperties().getName() + " Process Thread on " + destinationConnector.getChannel().getName() + " (" + chainProvider.getChannelId() + "), " + destinationConnector.getDestinationName() + " (" + metaDataId + ")"); - destinationConnector.process(dao, message, previousStatus); - } finally { - Thread.currentThread().setName(originalThreadName); + Status previousStatus = message.getStatus(); + if (previousStatus == Status.RECEIVED || previousStatus == Status.PENDING) { + observation = MessageTelemetry.start(MessageTelemetry.Stage.DESTINATION, message); + } + + try { + switch (message.getStatus()) { + case RECEIVED: + /* + * Only transform the message if we're going to be dispatching it in the + * main processing thread, or if the queue thread will not be handling + * transformation + */ + if (destinationConnector.willAttemptSend() || !destinationConnector.includeFilterTransformerInQueue()) { + destinationConnector.transform(dao, message, previousStatus, true); + + // If the message status is QUEUED, send it to the destination connector + if (message.getStatus() == Status.QUEUED) { + String originalThreadName = Thread.currentThread().getName(); + try { + Thread.currentThread().setName(destinationConnector.getConnectorProperties().getName() + " Process Thread on " + destinationConnector.getChannel().getName() + " (" + chainProvider.getChannelId() + "), " + destinationConnector.getDestinationName() + " (" + metaDataId + ")"); + destinationConnector.process(dao, message, previousStatus); + } finally { + Thread.currentThread().setName(originalThreadName); + } + } else if (message.getStatus() == Status.ERROR && message.getSent() == null) { + // If an error occurred in the filter/transformer, don't proceed with the rest of the chain + stopChain = true; } - } else if (message.getStatus() == Status.ERROR && message.getSent() == null) { - // If an error occurred in the filter/transformer, don't proceed with the rest of the chain - stopChain = true; + } else { + destinationConnector.updateQueuedStatus(dao, message, previousStatus); } - } else { - destinationConnector.updateQueuedStatus(dao, message, previousStatus); - } - break; + break; - case PENDING: - chainProvider.getDestinationConnectors().get(metaDataId).processPendingConnectorMessage(dao, message); - break; + case PENDING: + chainProvider.getDestinationConnectors().get(metaDataId).processPendingConnectorMessage(dao, message); + break; - case SENT: - break; + case SENT: + break; - default: - // the status should never be anything but one of the above statuses, but in case it's not, log an error - logger.error("Received a message with an invalid status in channel " + chainProvider.getChannelId() + "."); - break; - } - } catch (RuntimeException e) { // TODO: remove this catch since we can't determine an error code - // if an error occurred in processing the message through the current destination, then update the message status to ERROR and continue processing through the chain - logger.error("Error processing destination " + chainProvider.getDestinationConnectors().get(metaDataId).getDestinationName() + " for channel " + chainProvider.getChannelId() + ".", e); - stopChain = true; - dao.rollback(); - message.setStatus(Status.ERROR); - message.setProcessingError(e.toString()); - dao.updateStatus(message, previousStatus); - // Insert errors if necessary - if (StringUtils.isNotBlank(message.getProcessingError())) { - dao.updateErrors(message); + default: + // the status should never be anything but one of the above statuses, but in case it's not, log an error + logger.error("Received a message with an invalid status in channel " + chainProvider.getChannelId() + "."); + break; + } + } catch (RuntimeException e) { // TODO: remove this catch since we can't determine an error code + observationFailure = e; + // if an error occurred in processing the message through the current destination, then update the message status to ERROR and continue processing through the chain + logger.error("Error processing destination " + chainProvider.getDestinationConnectors().get(metaDataId).getDestinationName() + " for channel " + chainProvider.getChannelId() + ".", e); + stopChain = true; + dao.rollback(); + message.setStatus(Status.ERROR); + message.setProcessingError(e.toString()); + dao.updateStatus(message, previousStatus); + // Insert errors if necessary + if (StringUtils.isNotBlank(message.getProcessingError())) { + dao.updateErrors(message); + } } - } - // now that we're finished processing the current message, we can create the next message in the chain - if (nextMetaDataId != null && !stopChain) { - nextMessage = new ConnectorMessage(message.getChannelId(), message.getChannelName(), message.getMessageId(), nextMetaDataId, message.getServerId(), Calendar.getInstance(), Status.RECEIVED); + // now that we're finished processing the current message, we can create the next message in the chain + if (nextMetaDataId != null && !stopChain) { + nextMessage = new ConnectorMessage(message.getChannelId(), message.getChannelName(), message.getMessageId(), nextMetaDataId, message.getServerId(), Calendar.getInstance(), Status.RECEIVED); - DestinationConnector nextDestinationConnector = chainProvider.getDestinationConnectors().get(nextMetaDataId); - nextMessage.setConnectorName(nextDestinationConnector.getDestinationName()); - nextMessage.setChainId(chainProvider.getChainId()); - nextMessage.setOrderId(nextDestinationConnector.getOrderId()); + DestinationConnector nextDestinationConnector = chainProvider.getDestinationConnectors().get(nextMetaDataId); + nextMessage.setConnectorName(nextDestinationConnector.getDestinationName()); + nextMessage.setChainId(chainProvider.getChainId()); + nextMessage.setOrderId(nextDestinationConnector.getOrderId()); - // We don't create a new map here because the source map is read-only and thus won't ever be changed - nextMessage.setSourceMap(message.getSourceMap()); - nextMessage.setChannelMap(new HashMap(message.getChannelMap())); - nextMessage.setResponseMap(new HashMap(message.getResponseMap())); - nextMessage.setRaw(new MessageContent(message.getChannelId(), message.getMessageId(), nextMetaDataId, ContentType.RAW, message.getRaw().getContent(), nextDestinationConnector.getInboundDataType().getType(), message.getRaw().isEncrypted())); + // We don't create a new map here because the source map is read-only and thus won't ever be changed + nextMessage.setSourceMap(message.getSourceMap()); + MessageTelemetry.copyDispatchContext(message, nextMessage); + nextMessage.setChannelMap(new HashMap(message.getChannelMap())); + nextMessage.setResponseMap(new HashMap(message.getResponseMap())); + nextMessage.setRaw(new MessageContent(message.getChannelId(), message.getMessageId(), nextMetaDataId, ContentType.RAW, message.getRaw().getContent(), nextDestinationConnector.getInboundDataType().getType(), message.getRaw().isEncrypted())); - ThreadUtils.checkInterruptedStatus(); - dao.insertConnectorMessage(nextMessage, chainProvider.getStorageSettings().isStoreMaps(), true); - } + ThreadUtils.checkInterruptedStatus(); + dao.insertConnectorMessage(nextMessage, chainProvider.getStorageSettings().isStoreMaps(), true); + } - ThreadUtils.checkInterruptedStatus(); + ThreadUtils.checkInterruptedStatus(); - if (message.getStatus() != Status.QUEUED) { - dao.commit(chainProvider.getStorageSettings().isDurable()); - } else { - // Block other threads from reading from or modifying the destination queue until both the current commit and queue addition finishes - // Otherwise the same message could be sent multiple times. - synchronized (destinationConnector.getQueue()) { + if (message.getStatus() != Status.QUEUED) { dao.commit(chainProvider.getStorageSettings().isDurable()); - - if (message.getStatus() == Status.QUEUED) { - destinationConnector.getQueue().add(message); + } else { + // Block other threads from reading from or modifying the destination queue until both the current commit and queue addition finishes + // Otherwise the same message could be sent multiple times. + synchronized (destinationConnector.getQueue()) { + dao.commit(chainProvider.getStorageSettings().isDurable()); + + if (message.getStatus() == Status.QUEUED) { + destinationConnector.getQueue().add(message); + } } } + + messages.add(message); + } catch (RuntimeException e) { + // An exception caught at this point either occurred when attempting to handle an exception in the above try/catch, or when attempting to create the next destination's message, the thread cannot continue running + logger.error("Error processing destination " + chainProvider.getDestinationConnectors().get(metaDataId).getDestinationName() + " for channel " + chainProvider.getChannelId() + ".", e); + throw e; } - messages.add(message); - } catch (RuntimeException e) { - // An exception caught at this point either occurred when attempting to handle an exception in the above try/catch, or when attempting to create the next destination's message, the thread cannot continue running - logger.error("Error processing destination " + chainProvider.getDestinationConnectors().get(metaDataId).getDestinationName() + " for channel " + chainProvider.getChannelId() + ".", e); - throw e; + } catch (InterruptedException | RuntimeException | Error failure) { + observationFailure = MessageTelemetry.failure(observationFailure, failure); + throw failure; } finally { - dao.close(); + try { + dao.close(); + } catch (RuntimeException | Error failure) { + observationFailure = MessageTelemetry.failure(observationFailure, failure); + throw failure; + } finally { + MessageTelemetry.finish(observation, observationFailure); + } } // Set the next message in the loop @@ -210,4 +230,4 @@ private List doCall() throws InterruptedException { return messages; } -} \ No newline at end of file +} diff --git a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationConnector.java b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationConnector.java index dd74761264..6c3eadcc3a 100644 --- a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationConnector.java +++ b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/DestinationConnector.java @@ -633,218 +633,248 @@ public void run() { if (connectorMessage != null) { boolean exceptionCaught = false; + MessageTelemetry.Observation observation = null; + Throwable observationFailure = null; try { - /* - * If the last message id is equal to the current message id, then the - * message was not successfully sent and is being retried, so wait the retry - * interval. - * - * If the last message id is greater than the current message id, then some - * message was not successful, message rotation is on, and the queue is back - * to the oldest message, so wait the retry interval. - */ - if (connectorMessage.isAttemptedFirst() || lastMessageId != null && (lastMessageId == connectorMessage.getMessageId() || (queue.isRotate() && lastMessageId > connectorMessage.getMessageId() && queue.hasBeenRotated()))) { - try { - waitingRetryInterval.set(true); - Thread.sleep(retryIntervalMillis); - } finally { - synchronized (waitingRetryInterval) { - waitingRetryInterval.set(false); + try { + observation = MessageTelemetry.start(MessageTelemetry.Stage.DESTINATION, connectorMessage); + /* + * If the last message id is equal to the current message id, then the + * message was not successfully sent and is being retried, so wait the retry + * interval. + * + * If the last message id is greater than the current message id, then some + * message was not successful, message rotation is on, and the queue is back + * to the oldest message, so wait the retry interval. + */ + if (connectorMessage.isAttemptedFirst() || lastMessageId != null && (lastMessageId == connectorMessage.getMessageId() || (queue.isRotate() && lastMessageId > connectorMessage.getMessageId() && queue.hasBeenRotated()))) { + try { + waitingRetryInterval.set(true); + Thread.sleep(retryIntervalMillis); + } finally { + synchronized (waitingRetryInterval) { + waitingRetryInterval.set(false); + } } + + connectorMessage.setAttemptedFirst(false); } - connectorMessage.setAttemptedFirst(false); - } + lastMessageId = connectorMessage.getMessageId(); - lastMessageId = connectorMessage.getMessageId(); + dao = daoFactory.getDao(); + Status previousStatus = connectorMessage.getStatus(); - dao = daoFactory.getDao(); - Status previousStatus = connectorMessage.getStatus(); + Class connectorPropertiesClass = getConnectorProperties().getClass(); + Class serializedPropertiesClass = null; - Class connectorPropertiesClass = getConnectorProperties().getClass(); - Class serializedPropertiesClass = null; + ConnectorProperties connectorProperties = null; - ConnectorProperties connectorProperties = null; + /* + * If we're not regenerating connector properties, use the serialized sent + * content from the database. It's possible that the channel had Regenerate + * Template and Include Filter/Transformer enabled at one point, and then + * was disabled later, so we also have to make sure the sent content exists. + */ + if (!destinationConnectorProperties.isRegenerateTemplate() && connectorMessage.getSent() != null) { + // Attempt to get the sent properties from the in-memory cache. If it doesn't exist, deserialize from the actual sent content. + connectorProperties = connectorMessage.getSentProperties(); + if (connectorProperties == null) { + connectorProperties = serializer.deserialize(connectorMessage.getSent().getContent(), ConnectorProperties.class); + connectorMessage.setSentProperties(connectorProperties); + } - /* - * If we're not regenerating connector properties, use the serialized sent - * content from the database. It's possible that the channel had Regenerate - * Template and Include Filter/Transformer enabled at one point, and then - * was disabled later, so we also have to make sure the sent content exists. - */ - if (!destinationConnectorProperties.isRegenerateTemplate() && connectorMessage.getSent() != null) { - // Attempt to get the sent properties from the in-memory cache. If it doesn't exist, deserialize from the actual sent content. - connectorProperties = connectorMessage.getSentProperties(); - if (connectorProperties == null) { - connectorProperties = serializer.deserialize(connectorMessage.getSent().getContent(), ConnectorProperties.class); - connectorMessage.setSentProperties(connectorProperties); + serializedPropertiesClass = connectorProperties.getClass(); + } else { + connectorProperties = ((DestinationConnectorPropertiesInterface) getConnectorProperties()).clone(); } - serializedPropertiesClass = connectorProperties.getClass(); - } else { - connectorProperties = ((DestinationConnectorPropertiesInterface) getConnectorProperties()).clone(); - } - - /* - * Verify that the connector properties stored in the connector message - * match the properties from the current connector. Otherwise the connector - * type has changed and the message will be set to errored. If we're - * regenerating the connector properties then it doesn't matter. - */ - if (connectorMessage.getSent() == null || destinationConnectorProperties.isRegenerateTemplate() || serializedPropertiesClass == connectorPropertiesClass) { - ThreadUtils.checkInterruptedStatus(); - /* - * If a historical queued message has not yet been transformed and the - * current queue settings do not include the filter/transformer, force - * the message to ERROR. + * Verify that the connector properties stored in the connector message + * match the properties from the current connector. Otherwise the connector + * type has changed and the message will be set to errored. If we're + * regenerating the connector properties then it doesn't matter. */ - if (connectorMessage.getSent() == null && !includeFilterTransformerInQueue()) { - connectorMessage.setStatus(Status.ERROR); - connectorMessage.setProcessingError("Queued message has not yet been transformed, and Include Filter/Transformer is currently disabled."); + if (connectorMessage.getSent() == null || destinationConnectorProperties.isRegenerateTemplate() || serializedPropertiesClass == connectorPropertiesClass) { + ThreadUtils.checkInterruptedStatus(); - dao.updateStatus(connectorMessage, previousStatus); - dao.updateErrors(connectorMessage); - } else { - if (includeFilterTransformerInQueue()) { - transform(dao, connectorMessage, previousStatus, connectorMessage.getSent() == null); - } + /* + * If a historical queued message has not yet been transformed and the + * current queue settings do not include the filter/transformer, force + * the message to ERROR. + */ + if (connectorMessage.getSent() == null && !includeFilterTransformerInQueue()) { + connectorMessage.setStatus(Status.ERROR); + connectorMessage.setProcessingError("Queued message has not yet been transformed, and Include Filter/Transformer is currently disabled."); + + dao.updateStatus(connectorMessage, previousStatus); + dao.updateErrors(connectorMessage); + } else { + if (includeFilterTransformerInQueue()) { + transform(dao, connectorMessage, previousStatus, connectorMessage.getSent() == null); + } - if (connectorMessage.getStatus() == Status.QUEUED) { - /* - * Replace the connector properties if necessary. Again for - * historical queue reasons, we need to check whether the sent - * content exists. - */ - if (connectorMessage.getSent() == null || destinationConnectorProperties.isRegenerateTemplate()) { - replaceConnectorProperties(connectorProperties, connectorMessage); - MessageContent sentContent = getSentContent(connectorMessage, connectorProperties); - connectorMessage.setSent(sentContent); - - if (sentContent != null && storageSettings.isStoreSent()) { - ThreadUtils.checkInterruptedStatus(); - dao.storeMessageContent(sentContent); + if (connectorMessage.getStatus() == Status.QUEUED) { + /* + * Replace the connector properties if necessary. Again for + * historical queue reasons, we need to check whether the sent + * content exists. + */ + if (connectorMessage.getSent() == null || destinationConnectorProperties.isRegenerateTemplate()) { + replaceConnectorProperties(connectorProperties, connectorMessage); + MessageContent sentContent = getSentContent(connectorMessage, connectorProperties); + connectorMessage.setSent(sentContent); + + if (sentContent != null && storageSettings.isStoreSent()) { + ThreadUtils.checkInterruptedStatus(); + dao.storeMessageContent(sentContent); + } } - } - Response response = handleSend(connectorProperties, connectorMessage); - connectorMessage.setSendAttempts(connectorMessage.getSendAttempts() + 1); + Response response = handleSend(connectorProperties, connectorMessage); + connectorMessage.setSendAttempts(connectorMessage.getSendAttempts() + 1); - if (response == null) { - throw new RuntimeException("Received null response from destination " + destinationName + "."); - } - response.fixStatus(isQueueEnabled()); + if (response == null) { + throw new RuntimeException("Received null response from destination " + destinationName + "."); + } + response.fixStatus(isQueueEnabled()); - afterSend(dao, connectorMessage, response, previousStatus); + afterSend(dao, connectorMessage, response, previousStatus); + } } - } - } else { - connectorMessage.setStatus(Status.ERROR); - connectorMessage.setProcessingError("Mismatched connector properties detected in queued message. The connector type may have changed since the message was queued.\nFOUND: " + serializedPropertiesClass.getSimpleName() + "\nEXPECTED: " + connectorPropertiesClass.getSimpleName()); + } else { + connectorMessage.setStatus(Status.ERROR); + connectorMessage.setProcessingError("Mismatched connector properties detected in queued message. The connector type may have changed since the message was queued.\nFOUND: " + serializedPropertiesClass.getSimpleName() + "\nEXPECTED: " + connectorPropertiesClass.getSimpleName()); - dao.updateStatus(connectorMessage, previousStatus); - dao.updateErrors(connectorMessage); - } + dao.updateStatus(connectorMessage, previousStatus); + dao.updateErrors(connectorMessage); + } - /* - * If we're about to commit a non-QUEUED status, we first need to obtain a - * read lock from the queue. This is done so that if something else - * invalidates the queue at the same time, we don't incorrectly decrement - * the size during the release. - */ - if (connectorMessage.getStatus() != Status.QUEUED) { - Lock lock = queue.getStatusUpdateLock(); - lock.lock(); - statusUpdateLock = lock; - } + /* + * If we're about to commit a non-QUEUED status, we first need to obtain a + * read lock from the queue. This is done so that if something else + * invalidates the queue at the same time, we don't incorrectly decrement + * the size during the release. + */ + if (connectorMessage.getStatus() != Status.QUEUED) { + Lock lock = queue.getStatusUpdateLock(); + lock.lock(); + statusUpdateLock = lock; + } - ThreadUtils.checkInterruptedStatus(); - dao.commit(storageSettings.isDurable()); - commitSuccess = true; + ThreadUtils.checkInterruptedStatus(); + dao.commit(storageSettings.isDurable()); + commitSuccess = true; - // Only actually attempt to remove content if the status is SENT - if (connectorMessage.getStatus().isCompleted()) { - try { - channel.removeContent(dao, null, lastMessageId, true, true); - } catch (RuntimeException e) { - /* - * The connector message itself processed successfully, only the - * remove content operation failed. In this case just give up and - * log an error. - */ - logger.error("Error removing content for message " + lastMessageId + " for channel " + channel.getName() + " (" + channel.getChannelId() + ") on destination " + destinationName + ". This error is expected if the message was manually removed from the queue.", e); + // Only actually attempt to remove content if the status is SENT + if (connectorMessage.getStatus().isCompleted()) { + try { + channel.removeContent(dao, null, lastMessageId, true, true); + } catch (RuntimeException e) { + observationFailure = e; + /* + * The connector message itself processed successfully, only the + * remove content operation failed. In this case just give up and + * log an error. + */ + logger.error("Error removing content for message " + lastMessageId + " for channel " + channel.getName() + " (" + channel.getChannelId() + ") on destination " + destinationName + ". This error is expected if the message was manually removed from the queue.", e); + } } + } catch (RuntimeException e) { + observationFailure = e; + logger.error("Error processing queued " + (connectorMessage != null ? connectorMessage.toString() : "message (null)") + " for channel " + channel.getName() + " (" + channel.getChannelId() + ") on destination " + destinationName + ". This error is expected if the message was manually removed from the queue.", e); + /* + * Invalidate the queue's buffer if any errors occurred. If the message + * being processed by the queue was deleted, this will prevent the queue + * from trying to process that message repeatedly. Since multiple + * queues/threads may need to do this as well, we do not reset the queue's + * maps of checked in or deleted messages. + */ + exceptionCaught = true; + } catch (InterruptedException e) { + observationFailure = e; + // Stop this thread if it was halted + return; + } catch (Throwable t) { + observationFailure = t; + // Send a different error message to the server log, but still invalidate the queue buffer + logger.error("Error processing queued " + (connectorMessage != null ? connectorMessage.toString() : "message (null)") + " for channel " + channel.getName() + " (" + channel.getChannelId() + ") on destination " + destinationName + ".", t); + getChannel().getEventDispatcher().dispatchEvent(new ErrorEvent(getChannelId(), getMetaDataId(), connectorMessage != null ? connectorMessage.getMessageId() : null, ErrorEventType.DESTINATION_CONNECTOR, getDestinationName(), getConnectorProperties().getName(), t.getMessage(), t)); + exceptionCaught = true; } - } catch (RuntimeException e) { - logger.error("Error processing queued " + (connectorMessage != null ? connectorMessage.toString() : "message (null)") + " for channel " + channel.getName() + " (" + channel.getChannelId() + ") on destination " + destinationName + ". This error is expected if the message was manually removed from the queue.", e); - /* - * Invalidate the queue's buffer if any errors occurred. If the message - * being processed by the queue was deleted, this will prevent the queue - * from trying to process that message repeatedly. Since multiple - * queues/threads may need to do this as well, we do not reset the queue's - * maps of checked in or deleted messages. - */ - exceptionCaught = true; - } catch (InterruptedException e) { - // Stop this thread if it was halted - return; - } catch (Throwable t) { - // Send a different error message to the server log, but still invalidate the queue buffer - logger.error("Error processing queued " + (connectorMessage != null ? connectorMessage.toString() : "message (null)") + " for channel " + channel.getName() + " (" + channel.getChannelId() + ") on destination " + destinationName + ".", t); - getChannel().getEventDispatcher().dispatchEvent(new ErrorEvent(getChannelId(), getMetaDataId(), connectorMessage != null ? connectorMessage.getMessageId() : null, ErrorEventType.DESTINATION_CONNECTOR, getDestinationName(), getConnectorProperties().getName(), t.getMessage(), t)); - exceptionCaught = true; + } catch (RuntimeException | Error failure) { + observationFailure = MessageTelemetry.failure(observationFailure, failure); + throw failure; } finally { - if (dao != null) { - if (!commitSuccess) { - try { - dao.rollback(); - } catch (Exception e) {} + try { + if (dao != null) { + if (!commitSuccess) { + try { + dao.rollback(); + } catch (Exception e) {} + } + dao.close(); } - dao.close(); - } - /* - * We always want to release the message if it's done (obviously). - */ - if (exceptionCaught) { /* - * If an runtime exception was caught, we can't guarantee whether that - * message was deleted or is still in the database. When it is released, - * the message will be removed from the in-memory queue. However we need - * to invalidate the queue before allowing any other threads to be able - * to access it in case the message is still in the database. + * We always want to release the message if it's done (obviously). */ - canAcquire = true; - synchronized (queue) { + if (exceptionCaught) { + /* + * If an runtime exception was caught, we can't guarantee whether that + * message was deleted or is still in the database. When it is released, + * the message will be removed from the in-memory queue. However we need + * to invalidate the queue before allowing any other threads to be able + * to access it in case the message is still in the database. + */ + canAcquire = true; + synchronized (queue) { + queue.release(connectorMessage, true); + + // Release the read lock now before calling invalidate + if (statusUpdateLock != null) { + statusUpdateLock.unlock(); + statusUpdateLock = null; + } + + queue.invalidate(true, false); + } + } else if (connectorMessage.getStatus() != Status.QUEUED) { + canAcquire = true; queue.release(connectorMessage, true); + } else if (destinationConnectorProperties.isRotate()) { + canAcquire = true; + queue.release(connectorMessage, false); + } else { + /* + * If the message is still queued, no exception occurred, and queue + * rotation is disabled, we still want to force the queue to re-acquire + * a message if it has been marked as deleted by another process. + */ + canAcquire = queue.releaseIfDeleted(connectorMessage); + } - // Release the read lock now before calling invalidate + // Always release the read lock if we obtained it + if (statusUpdateLock != null) { + statusUpdateLock.unlock(); + statusUpdateLock = null; + } + } catch (RuntimeException | Error failure) { + observationFailure = MessageTelemetry.failure(observationFailure, failure); + throw failure; + } finally { + try { + // Cleanup failure may have skipped the normal status-lock release. if (statusUpdateLock != null) { statusUpdateLock.unlock(); statusUpdateLock = null; } - - queue.invalidate(true, false); + } catch (RuntimeException | Error failure) { + observationFailure = MessageTelemetry.failure(observationFailure, failure); + throw failure; + } finally { + MessageTelemetry.finish(observation, observationFailure); } - } else if (connectorMessage.getStatus() != Status.QUEUED) { - canAcquire = true; - queue.release(connectorMessage, true); - } else if (destinationConnectorProperties.isRotate()) { - canAcquire = true; - queue.release(connectorMessage, false); - } else { - /* - * If the message is still queued, no exception occurred, and queue - * rotation is disabled, we still want to force the queue to re-acquire - * a message if it has been marked as deleted by another process. - */ - canAcquire = queue.releaseIfDeleted(connectorMessage); - } - - // Always release the read lock if we obtained it - if (statusUpdateLock != null) { - statusUpdateLock.unlock(); - statusUpdateLock = null; } } } else { @@ -895,6 +925,13 @@ public void run() { } private Response handleSend(ConnectorProperties connectorProperties, ConnectorMessage message) throws InterruptedException { + try (var observation = MessageTelemetry.start(MessageTelemetry.Stage.SEND, message)) { + try { Response response = sendMessage(connectorProperties, message); observation.status(response.getStatus()); return response; } + catch (InterruptedException | RuntimeException | Error failure) { observation.failed(failure); throw failure; } + } + } + + private Response sendMessage(ConnectorProperties connectorProperties, ConnectorMessage message) throws InterruptedException { message.setSendDate(Calendar.getInstance()); Response response; diff --git a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/FilterTransformerExecutor.java b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/FilterTransformerExecutor.java index 419ded2dc3..953820da00 100644 --- a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/FilterTransformerExecutor.java +++ b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/FilterTransformerExecutor.java @@ -63,6 +63,13 @@ public void setFilterTransformer(FilterTransformer filterTransformer) { * @throws InterruptedException */ public void processConnectorMessage(ConnectorMessage connectorMessage) throws InterruptedException, DonkeyException { + try (var observation = MessageTelemetry.start(MessageTelemetry.Stage.TRANSFORM, connectorMessage)) { + try { transformMessage(connectorMessage); } + catch (InterruptedException | DonkeyException | RuntimeException | Error failure) { observation.failed(failure); throw failure; } + } + } + + private void transformMessage(ConnectorMessage connectorMessage) throws InterruptedException, DonkeyException { ThreadUtils.checkInterruptedStatus(); String content; String encodedContent; diff --git a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/MessageTelemetry.java b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/MessageTelemetry.java new file mode 100644 index 0000000000..fd906e1b7c --- /dev/null +++ b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/MessageTelemetry.java @@ -0,0 +1,196 @@ +/* Published under the Mozilla Public License 2.0. */ +package com.mirth.connect.donkey.server.channel; + +import java.util.Objects; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.model.message.Status; +import org.apache.logging.log4j.LogManager; + +/** + * Optional in-process observer. Providers own telemetry; callbacks must not block or mutate + * message behavior. An observation belongs to its starting thread. A provider that fails during + * start/activation must restore any context it already attached before throwing. + */ +public final class MessageTelemetry { + public enum Stage { SOURCE, TRANSFORM, SEND, RESPONSE, DESTINATION } + public interface Observation extends AutoCloseable { + default void status(Status status) { } + default void failed(Throwable failure) { } + @Override void close(); + } + public interface Provider { + Observation start(Stage stage, ConnectorMessage message); + /** Static API capability, not a cached policy/admission decision. No live resources. */ + default boolean supportsDispatchContext() { return false; } + /** + * Optional immutable, bounded, content-free preparation proof. It must retain no message, + * application map, Throwable, Context, SDK/provider or lease. Persisted data stays in the + * existing map representation. Null opts this dispatch into the ordinary start path. + */ + default Object prepareDispatch(ConnectorMessage message, Map sourceMap) { + beforeStore(message, sourceMap); + return null; + } + /** Providers must recheck native coordinates, observed map values and current policy. */ + default Observation start(Stage stage, ConnectorMessage message, Object dispatchContext) { + return start(stage, message); + } + /** + * Before the source map becomes read-only and is first persisted. May add only private, + * content-free propagation data to sourceMap; preserve all application entries. This + * callback must open no span/scope or other resource requiring later cleanup. + */ + default void beforeStore(ConnectorMessage message, Map sourceMap) { } + /** + * Capture immutable context now, without opening a scope or allocating live resources. + * The supplier attaches it only if the task runs; its observation restores worker state. + * Null means no transfer is needed. Never capture a thread-bound open scope. + */ + default Supplier capture() { return null; } + } + private static final Observation NONE = () -> { }; + private static final AtomicReference CURRENT = new AtomicReference<>(); + private MessageTelemetry() { } + + /** + * Exactly one provider; the idempotent token detaches only this installation and never blocks + * traffic. Existing observations and captured context activations retain the old provider, + * which must allow them to finish safely after detach. New stage observations (including those + * inside captured tasks) select the current installation. This token does not shut down resources. + */ + public static AutoCloseable install(Provider provider) { + Registration registration = new Registration(Objects.requireNonNull(provider)); + if (!CURRENT.compareAndSet(null, registration)) throw new IllegalStateException("Message telemetry already installed"); + return () -> CURRENT.compareAndSet(registration, null); + } + public static Observation start(Stage stage, ConnectorMessage message) { + Registration registration = CURRENT.get(); + if (registration == null) return NONE; + return observe(registration, () -> registration.dispatchOwner == null + ? registration.provider.start(stage, message) + : registration.provider.start(stage, message, message == null ? null + : message.getTelemetryContext(registration.dispatchOwner))); + } + public static void beforeStore(ConnectorMessage message, Map sourceMap) { + Registration registration = CURRENT.get(); + if (registration == null) return; + if (registration.dispatchOwner == null || message == null) { + try { registration.provider.beforeStore(message, sourceMap); } + catch (Throwable failure) { registration.failed(failure); } + return; + } + Object reservation = null; + try { + reservation = message.reserveTelemetryContext(registration.dispatchOwner); + Object prepared = registration.provider.prepareDispatch(message, sourceMap); + message.completeTelemetryContext(reservation, prepared); + } catch (Throwable failure) { + if (reservation != null) message.completeTelemetryContext(reservation, null); + registration.failed(failure); + } + } + + /** Native dispatch copies only this installation's scalar proof, never a source message/SDK. */ + static void copyDispatchContext(ConnectorMessage source, ConnectorMessage destination) { + Registration registration = CURRENT.get(); + if (registration == null || registration.dispatchOwner == null || source == null || destination == null) return; + try { + Object value = source.getTelemetryContext(registration.dispatchOwner); + Object reservation = destination.reserveTelemetryContext(registration.dispatchOwner); + destination.completeTelemetryContext(reservation, value); + } catch (Throwable failure) { registration.failed(failure); } + } + private static Observation observe(Registration registration, Supplier start) { + try { + Observation observation = start.get(); + return observation == null ? NONE : new Observation() { + public void status(Status status) { try { observation.status(status); } catch (Throwable failure) { registration.failed(failure); } } + public void failed(Throwable cause) { + try { observation.failed(cause); } + catch (Throwable failure) { + if (fatal(cause) && fatal(failure)) { + if (cause != failure) { + try { cause.addSuppressed(failure); } + catch (Throwable ignored) { /* Preserve the original fatal even if suppression cannot allocate. */ } + } + } else registration.failed(failure); + } + } + public void close() { try { observation.close(); } catch (Throwable failure) { registration.failed(failure); } } + }; + } catch (Throwable failure) { registration.failed(failure); return NONE; } + } + private static boolean fatal(Throwable failure) { + return failure instanceof VirtualMachineError || failure instanceof ThreadDeath; + } + /** Retain the first observed engine fatal across later handling or cleanup failures. */ + static Throwable failure(Throwable previous, Throwable next) { + return fatal(previous) ? previous : next; + } + /** Finish after engine cleanup, including failures the engine deliberately catches. */ + static void finish(Observation observation, Throwable failure) { + Throwable problem = null; + try { + if (observation != null && failure != null) observation.failed(failure); + } catch (VirtualMachineError | ThreadDeath telemetryFailure) { problem = telemetryFailure; } + finally { + try { if (observation != null) observation.close(); } + catch (VirtualMachineError | ThreadDeath telemetryFailure) { + if (problem == null) problem = telemetryFailure; + else suppress(problem, telemetryFailure); + } + } + if (problem != null) { + if (fatal(failure)) suppress(failure, problem); + else if (problem instanceof VirtualMachineError) throw (VirtualMachineError) problem; + else throw (ThreadDeath) problem; + } + } + private static void suppress(Throwable primary, Throwable secondary) { + if (primary != secondary) { + try { primary.addSuppressed(secondary); } + catch (Throwable ignored) { /* Preserve the first fatal even if suppression cannot allocate. */ } + } + } + public static Callable wrap(Callable task) { + Objects.requireNonNull(task); + Registration registration = CURRENT.get(); + if (registration == null) return task; + try { + Supplier captured = registration.provider.capture(); + if (captured == null) return task; + return () -> { + try (Observation scope = observe(registration, captured)) { + return task.call(); + } + }; + } + catch (Throwable failure) { registration.failed(failure); return task; } + } + private static final class Registration { + final Provider provider; + // This key is deliberately not Registration: messages must not retain Provider/SDK owners. + final Object dispatchOwner; + final AtomicBoolean warned = new AtomicBoolean(); + Registration(Provider provider) { + this.provider = provider; + dispatchOwner = provider.supportsDispatchContext() ? new Object() : null; + } + void failed(Throwable failure) { + if (failure instanceof VirtualMachineError) throw (VirtualMachineError) failure; + if (failure instanceof ThreadDeath) throw (ThreadDeath) failure; + if (warned.compareAndSet(false, true)) { + try { + LogManager.getLogger(MessageTelemetry.class) + .warn("Message telemetry callback failed; further warnings for this installation are suppressed"); + } catch (VirtualMachineError | ThreadDeath fatal) { throw fatal; } + catch (Throwable loggingFailure) { /* A broken appender must not break message processing. */ } + } + } + } +} diff --git a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/ResponseTransformerExecutor.java b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/ResponseTransformerExecutor.java index a9c77af42d..4946871341 100644 --- a/donkey/src/main/java/com/mirth/connect/donkey/server/channel/ResponseTransformerExecutor.java +++ b/donkey/src/main/java/com/mirth/connect/donkey/server/channel/ResponseTransformerExecutor.java @@ -58,6 +58,13 @@ public void setResponseTransformer(ResponseTransformer responseTransformer) { } public void runResponseTransformer(DonkeyDao dao, ConnectorMessage connectorMessage, Response response, boolean queueEnabled, StorageSettings storageSettings, Serializer serializer) throws InterruptedException, DonkeyException { + try (var observation = MessageTelemetry.start(MessageTelemetry.Stage.RESPONSE, connectorMessage)) { + try { transformResponse(dao, connectorMessage, response, queueEnabled, storageSettings, serializer); observation.status(response.getStatus()); } + catch (InterruptedException | DonkeyException | RuntimeException | Error failure) { observation.failed(failure); throw failure; } + } + } + + private void transformResponse(DonkeyDao dao, ConnectorMessage connectorMessage, Response response, boolean queueEnabled, StorageSettings storageSettings, Serializer serializer) throws InterruptedException, DonkeyException { ThreadUtils.checkInterruptedStatus(); String processedResponseContent; diff --git a/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryAdversarialTest.java b/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryAdversarialTest.java new file mode 100644 index 0000000000..a68db9d0cb --- /dev/null +++ b/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryAdversarialTest.java @@ -0,0 +1,147 @@ +/* Published under the Mozilla Public License 2.0. */ +package com.mirth.connect.donkey.server.channel; + +import static org.junit.Assert.*; +import org.junit.Test; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.function.Supplier; +import java.util.*; +import com.mirth.connect.donkey.server.channel.*; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.*; +import com.mirth.connect.donkey.model.message.*; +import com.mirth.connect.donkey.model.DonkeyException; +import com.mirth.connect.donkey.test.util.*; + +public class MessageTelemetryAdversarialTest { + interface Action { void run() throws Throwable; } + static Throwable thrown(Action action) { + try { action.run(); return null; } catch (Throwable t) { return t; } + } + static ConnectorMessage message() { + ConnectorMessage message = new ConnectorMessage(); + message.setStatus(Status.RECEIVED); + message.setRaw(new MessageContent("review", 1L, 0, ContentType.RAW, "content", "RAW", false)); + return message; + } + @Test public void allOrdinaryCallbackSurfacesPreserveOriginalFailure() throws Exception { + for (int surface = 0; surface < 6; surface++) { + final int selected = surface; + for (Error callback : new Error[] {new AssertionError("sentinel"), new LinkageError("sentinel")}) { + AtomicInteger called = new AtomicInteger(), closed = new AtomicInteger(); + Provider provider = new Provider() { + Observation observation() { + return new Observation() { + public void status(Status s) { if (selected == 3) throw callback; } + public void failed(Throwable t) { if (selected == 4) throw callback; } + public void close() { closed.incrementAndGet(); if (selected == 5) throw callback; } + }; + } + public Observation start(Stage s, ConnectorMessage m) { if (selected == 0) throw callback; return observation(); } + public Supplier capture() { if (selected == 1) throw callback; return () -> { if (selected == 2) throw callback; return () -> { closed.incrementAndGet(); }; }; } + }; + try (AutoCloseable token = MessageTelemetry.install(provider)) { + for (Throwable original : new Throwable[] {new RuntimeException(), new AssertionError(), new OutOfMemoryError(), new ThreadDeath()}) { + Throwable actual = thrown(() -> MessageTelemetry.wrap(() -> { + called.incrementAndGet(); + try (Observation observation = MessageTelemetry.start(Stage.SOURCE, null)) { + observation.status(Status.ERROR); + observation.failed(original); + if (original instanceof Error) throw (Error) original; + throw (RuntimeException) original; + } + }).call()); + assertSame("surface " + selected, original, actual); + assertEquals(0, actual.getSuppressed().length); + } + assertEquals(4, called.get()); + assertTrue(closed.get() >= 4); + } + } + } + } + @Test public void fatalSurfacesPropagateAndCloseUsesJavaSuppression() throws Exception { + for (int surface = 0; surface < 6; surface++) { + final int selected = surface; + for (Error callback : new Error[] {new OutOfMemoryError("sentinel"), new ThreadDeath()}) { + AtomicInteger called = new AtomicInteger(), closed = new AtomicInteger(); + Provider provider = new Provider() { + Observation observation() { return new Observation() { + public void status(Status s) { if (selected == 3) throw callback; } + public void failed(Throwable cause) { if (selected == 4) throw callback; } + public void close() { closed.incrementAndGet(); if (selected == 5) throw callback; } + }; } + public Observation start(Stage s, ConnectorMessage m) { if (selected == 0) throw callback; return observation(); } + public Supplier capture() { if (selected == 1) throw callback; return () -> { if (selected == 2) throw callback; return () -> { closed.incrementAndGet(); }; }; } + }; + try (AutoCloseable token = MessageTelemetry.install(provider)) { + Throwable actual = thrown(() -> MessageTelemetry.wrap(() -> { + called.incrementAndGet(); + try (Observation observation = MessageTelemetry.start(Stage.SOURCE, null)) { + observation.status(Status.ERROR); + observation.failed(new RuntimeException("business")); + } + return "done"; + }).call()); + assertSame("surface " + selected, callback, actual); + assertEquals(selected == 1 || selected == 2 ? 0 : 1, called.get()); + } + } + } + Error business = new OutOfMemoryError("business"), cleanup = new ThreadDeath(); + try (AutoCloseable token = MessageTelemetry.install((s, m) -> () -> { throw cleanup; })) { + Throwable actual = thrown(() -> { try (Observation o = MessageTelemetry.start(Stage.SOURCE, null)) { throw business; } }); + assertSame(business, actual); + assertArrayEquals(new Throwable[] {cleanup}, actual.getSuppressed()); + } + } + @Test public void capturedActivationUsesOldProviderButNewStartsUseReplacement() throws Exception { + AtomicInteger oldStarts = new AtomicInteger(), newStarts = new AtomicInteger(), oldActivations = new AtomicInteger(); + AutoCloseable old = MessageTelemetry.install(new Provider() { + public Observation start(Stage stage, ConnectorMessage message) { oldStarts.incrementAndGet(); return null; } + public Supplier capture() { return () -> { oldActivations.incrementAndGet(); return () -> {}; }; } + }); + Callable captured = MessageTelemetry.wrap(() -> { MessageTelemetry.start(Stage.TRANSFORM, null).close(); return null; }); + old.close(); + try (AutoCloseable replacement = MessageTelemetry.install((s, m) -> { newStarts.incrementAndGet(); return null; })) { + captured.call(); old.close(); captured.call(); + assertEquals(0, oldStarts.get()); assertEquals(2, newStarts.get()); assertEquals(2, oldActivations.get()); + } + } + @Test public void concurrentInstallHasSingleWinnerAndStaleCloseIsHarmless() throws Exception { + ExecutorService workers = Executors.newFixedThreadPool(8); + CountDownLatch start = new CountDownLatch(1); + List winners = new CopyOnWriteArrayList<>(); + List> futures = new ArrayList<>(); + try { + for (int i = 0; i < 8; i++) futures.add(workers.submit(() -> { start.await(); try { winners.add(MessageTelemetry.install((s, m) -> null)); } catch (IllegalStateException expected) {} return null; })); + start.countDown(); + for (Future future : futures) future.get(5, TimeUnit.SECONDS); + assertEquals(1, winners.size()); + winners.get(0).close(); + AtomicInteger starts = new AtomicInteger(); + try (AutoCloseable replacement = MessageTelemetry.install((s, m) -> { starts.incrementAndGet(); return null; })) { + for (int i = 0; i < 1000; i++) winners.get(0).close(); + MessageTelemetry.start(Stage.SOURCE, null).close(); assertEquals(1, starts.get()); + } + } finally { for (AutoCloseable winner : winners) winner.close(); workers.shutdownNow(); assertTrue(workers.awaitTermination(5, TimeUnit.SECONDS)); } + } + @Test public void originalFatalEngineFailureMustSurviveFatalFailedCallback() throws Exception { + for (Error business : new Error[] {new OutOfMemoryError("engine-fatal"), new ThreadDeath()}) { + Error callback = new OutOfMemoryError("callback-fatal"); + AtomicInteger closed = new AtomicInteger(); + FilterTransformerExecutor transform = TestUtils.createDefaultFilterTransformerExecutor(); + transform.setFilterTransformer(new TestFilterTransformer() { + public FilterTransformerResult doFilterTransform(ConnectorMessage m) { throw business; } + }); + try (AutoCloseable token = MessageTelemetry.install((s, m) -> new Observation() { + public void failed(Throwable failure) { assertSame(business, failure); throw callback; } + public void close() { closed.incrementAndGet(); } + })) { + Throwable actual = thrown(() -> transform.processConnectorMessage(message())); + assertEquals(1, closed.get()); + assertSame("fatal business error was replaced; actual=" + actual + "; suppressed=" + Arrays.toString(actual.getSuppressed()), business, actual); + } + } + } +} diff --git a/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryDispatchTest.java b/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryDispatchTest.java new file mode 100644 index 0000000000..12c637b20f --- /dev/null +++ b/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryDispatchTest.java @@ -0,0 +1,167 @@ +/* MPL-2.0. Isolated additive native dispatch-context protocol controls. */ +package com.mirth.connect.donkey.server.channel; + +import static org.junit.Assert.*; +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.model.message.Status; +import com.mirth.connect.donkey.util.xstream.XStreamSerializer; +import java.io.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import org.junit.Test; + +public class MessageTelemetryDispatchTest { + static final MessageTelemetry.Observation NONE = () -> {}; + static ConnectorMessage message(long id) { return new ConnectorMessage("channel", "name", id, 0, "server", Calendar.getInstance(), Status.RECEIVED); } + static class Provider implements MessageTelemetry.Provider { + final List seen = new CopyOnWriteArrayList<>(); + final AtomicInteger legacyStarts = new AtomicInteger(), legacyPrepares = new AtomicInteger(), prepares = new AtomicInteger(); + volatile Object prepared = new Object(); + public boolean supportsDispatchContext() { return true; } + public MessageTelemetry.Observation start(MessageTelemetry.Stage stage, ConnectorMessage message) { legacyStarts.incrementAndGet(); return NONE; } + public void beforeStore(ConnectorMessage message, Map map) { legacyPrepares.incrementAndGet(); } + public Object prepareDispatch(ConnectorMessage message, Map map) { prepares.incrementAndGet(); return prepared; } + public MessageTelemetry.Observation start(MessageTelemetry.Stage stage, ConnectorMessage message, Object proof) { seen.add(proof == null ? "absent" : proof); return NONE; } + } + + @Test public void ownerSlotIsOpaqueTransientAndCompareCompleted() { + var message=message(1); var owner=new Object(); var other=new Object(); var value=new Object(); + Object first=message.reserveTelemetryContext(owner); assertNull(message.getTelemetryContext(owner)); + Object second=message.reserveTelemetryContext(owner); assertFalse(message.completeTelemetryContext(first,value)); + assertFalse(message.completeTelemetryContext(new Object(),value)); assertTrue(message.completeTelemetryContext(second,value)); + assertSame(value,message.getTelemetryContext(owner)); assertNull(message.getTelemetryContext(other)); assertNull(message.getTelemetryContext(null)); + assertFalse(message.completeTelemetryContext(second,null)); assertSame(value,message.getTelemetryContext(owner)); + var replacement=message.reserveTelemetryContext(other); assertNull(message.getTelemetryContext(owner)); + assertTrue(message.completeTelemetryContext(replacement,null)); assertNull(message.getTelemetryContext(other)); + } + + @Test public void unchangedLegacyProviderStillUsesOnlyItsOriginalCallbacks() throws Exception { + var provider=new Provider() { public boolean supportsDispatchContext(){return false;} }; + try(var registration=MessageTelemetry.install(provider)) { + var message=message(1);MessageTelemetry.beforeStore(message,message.getSourceMap());MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close(); + assertEquals(1,provider.legacyPrepares.get()); assertEquals(1,provider.legacyStarts.get());assertEquals(0,provider.prepares.get()); assertTrue(provider.seen.isEmpty()); + } + } + + @Test public void preparedContextReachesOnlyTheCurrentRegistrationAndCopyTarget() throws Exception { + var provider=new Provider();var source=message(1);var target=message(1); + try(var registration=MessageTelemetry.install(provider)) { + MessageTelemetry.beforeStore(source,source.getSourceMap());MessageTelemetry.copyDispatchContext(source,target); + MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,source).close();MessageTelemetry.start(MessageTelemetry.Stage.DESTINATION,target).close(); + assertEquals(List.of(provider.prepared,provider.prepared),provider.seen);assertEquals(0,provider.legacyStarts.get()); + } + provider.seen.clear(); + try(var next=MessageTelemetry.install(provider)) { + MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,source).close();MessageTelemetry.start(MessageTelemetry.Stage.DESTINATION,target).close(); + assertEquals(List.of("absent","absent"),provider.seen); + } + } + + @Test public void arbitraryApplicationSlotCannotForgeRegistrationOwnership() throws Exception { + var provider=new Provider();var message=message(1); + try(var registration=MessageTelemetry.install(provider)) { + MessageTelemetry.beforeStore(message,message.getSourceMap()); + Object forged=message.reserveTelemetryContext(new Object());message.completeTelemetryContext(forged,provider.prepared); + MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close();assertEquals(List.of("absent"),provider.seen); + } + } + + @Test public void copyingAnUnpreparedSourceCannotLeaveAStaleDestinationProof() throws Exception { + var provider=new Provider();var source=message(1);var target=message(1); + try(var registration=MessageTelemetry.install(provider)) { + MessageTelemetry.beforeStore(target,target.getSourceMap()); + MessageTelemetry.copyDispatchContext(source,target); + MessageTelemetry.start(MessageTelemetry.Stage.DESTINATION,target).close();assertEquals(List.of("absent"),provider.seen); + } + } + + @Test public void capabilityDeclarationFailureDoesNotLeaveARegistrationInstalled() throws Exception { + var sentinel=new IllegalStateException("capability declaration"); + var broken=new Provider(){public boolean supportsDispatchContext(){throw sentinel;}}; + try{MessageTelemetry.install(broken);fail("declaration failure required");}catch(IllegalStateException actual){assertSame(sentinel,actual);} + var next=new Provider();try(var registration=MessageTelemetry.install(next)){MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message(1)).close();assertEquals(List.of("absent"),next.seen);} + } + + @Test public void oldObservationKeepsItsOwnerAfterReplacementEvenWithDispatchState() throws Exception { + var closes=new AtomicInteger();var first=new Provider(){public MessageTelemetry.Observation start(MessageTelemetry.Stage stage,ConnectorMessage m,Object proof){assertSame(prepared,proof);return closes::incrementAndGet;}}; + var next=new Provider();var message=message(1);AutoCloseable a=MessageTelemetry.install(first),b=null; + try { + MessageTelemetry.beforeStore(message,message.getSourceMap());var observation=MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message); + a.close();b=MessageTelemetry.install(next);MessageTelemetry.beforeStore(message,message.getSourceMap());observation.close(); + assertEquals(1,closes.get());MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close();assertEquals(List.of(next.prepared),next.seen); + }finally{a.close();if(b!=null)b.close();} + } + + @Test public void nativeSlotKeyIsNotAProviderOrRegistrationReference() throws Exception { + var provider=new Provider();var message=message(1); + try(var registration=MessageTelemetry.install(provider)) { + MessageTelemetry.beforeStore(message,message.getSourceMap()); + var field=ConnectorMessage.class.getDeclaredField("telemetrySlot");field.setAccessible(true);Object slot=field.get(message); + var key=slot.getClass().getDeclaredField("owner");key.setAccessible(true);assertEquals(Object.class,key.get(slot).getClass()); + assertTrue(java.lang.reflect.Modifier.isTransient(field.getModifiers())); + } + } + + @Test public void failedPreparationCannotReuseAnEarlierProof() throws Exception { + var fail=new AtomicBoolean();var provider=new Provider(){public Object prepareDispatch(ConnectorMessage m,Map map){if(fail.get())throw new IllegalStateException("private failure");return super.prepareDispatch(m,map);}}; + var message=message(1); + try(var registration=MessageTelemetry.install(provider)) { + MessageTelemetry.beforeStore(message,message.getSourceMap());fail.set(true);MessageTelemetry.beforeStore(message,message.getSourceMap()); + MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close();assertEquals(List.of("absent"),provider.seen); + } + } + + @Test public void fatalPreparationRetiresOnlyItsReservationAndPropagatesIdentity() throws Exception { + var sentinel=new OutOfMemoryError("synthetic preparation fatal");var provider=new Provider(){public Object prepareDispatch(ConnectorMessage m,Map map){throw sentinel;}}; + var message=message(1); + try(var registration=MessageTelemetry.install(provider)) { + try{MessageTelemetry.beforeStore(message,message.getSourceMap());fail("fatal required");}catch(OutOfMemoryError actual){assertSame(sentinel,actual);} + MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close();assertEquals(List.of("absent"),provider.seen); + } + } + + @Test public void reentrantNewerPreparationWinsEvenIfOuterFails() throws Exception { + for(boolean failOuter:new boolean[]{false,true}) { + var inner=new Object();var outer=new Object();var depth=new AtomicInteger(); + var provider=new Provider(){public Object prepareDispatch(ConnectorMessage m,Map map){ + if(depth.getAndIncrement()==0){MessageTelemetry.beforeStore(m,map);if(failOuter)throw new IllegalStateException("outer failed");return outer;}return inner; + }}; + var message=message(1); + try(var registration=MessageTelemetry.install(provider)) { + MessageTelemetry.beforeStore(message,message.getSourceMap());MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close();assertEquals(List.of(inner),provider.seen); + } + } + } + + @Test public void inFlightOldPreparationCannotGrantTheReplacementProviderItsProof() throws Exception { + var entered=new CountDownLatch(1);var release=new CountDownLatch(1); + var old=new Provider(){public Object prepareDispatch(ConnectorMessage m,Map map){entered.countDown();try{assertTrue(release.await(2,TimeUnit.SECONDS));}catch(InterruptedException e){Thread.currentThread().interrupt();throw new AssertionError(e);}return prepared;}}; + var next=new Provider();var message=message(1);var worker=Executors.newSingleThreadExecutor(); + AutoCloseable first=MessageTelemetry.install(old),second=null; + try { + var future=worker.submit(()->MessageTelemetry.beforeStore(message,message.getSourceMap()));assertTrue(entered.await(2,TimeUnit.SECONDS));first.close();second=MessageTelemetry.install(next); + MessageTelemetry.beforeStore(message,message.getSourceMap());release.countDown();future.get(2,TimeUnit.SECONDS); + MessageTelemetry.start(MessageTelemetry.Stage.SOURCE,message).close();assertEquals(List.of(next.prepared),next.seen); + }finally{release.countDown();first.close();if(second!=null)second.close();worker.shutdownNow();assertTrue(worker.awaitTermination(2,TimeUnit.SECONDS));} + } + + @Test public void slotOwnerAndPayloadCannotMixAcrossConcurrentReplacements() throws Exception { + var message=message(1);var a=new Object();var b=new Object();var worker=Executors.newFixedThreadPool(2); + try { + var first=worker.submit(()->{for(int i=0;i<10000;i++){Object slot=message.reserveTelemetryContext(a);message.completeTelemetryContext(slot,a);Object read=message.getTelemetryContext(a);assertTrue(read==null||read==a);}}); + var second=worker.submit(()->{for(int i=0;i<10000;i++){Object slot=message.reserveTelemetryContext(b);message.completeTelemetryContext(slot,b);Object read=message.getTelemetryContext(b);assertTrue(read==null||read==b);}}); + first.get(3,TimeUnit.SECONDS);second.get(3,TimeUnit.SECONDS); + }finally{worker.shutdownNow();assertTrue(worker.awaitTermination(2,TimeUnit.SECONDS));} + } + + @Test public void dispatchContextDoesNotEnterJavaOrXmlSerialization() throws Exception { + var message=message(1);var serializer=new XStreamSerializer();String original=serializer.serialize(message); + var key=new Object();message.completeTelemetryContext(message.reserveTelemetryContext(key),new Object()); + assertEquals(original,serializer.serialize(message));assertFalse(original.contains("telemetrySlot")); + assertEquals(8556587410415698618L,ObjectStreamClass.lookup(ConnectorMessage.class).getSerialVersionUID()); + var bytes=new ByteArrayOutputStream();try(var output=new ObjectOutputStream(bytes)){output.writeObject(message);} + ConnectorMessage restored;try(var input=new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))){restored=(ConnectorMessage)input.readObject();} + assertNull(restored.getTelemetryContext(key));assertEquals(message.getMessageId(),restored.getMessageId());assertEquals(message.getSourceMap(),restored.getSourceMap()); + } +} diff --git a/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryTest.java b/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryTest.java new file mode 100644 index 0000000000..9fb0c448ef --- /dev/null +++ b/donkey/src/test/java/com/mirth/connect/donkey/server/channel/MessageTelemetryTest.java @@ -0,0 +1,309 @@ +/* Published under the Mozilla Public License 2.0. */ +package com.mirth.connect.donkey.server.channel; + +import static org.junit.Assert.*; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.HashMap; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import org.junit.After; +import org.junit.Test; + +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.model.message.Status; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.Observation; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.Provider; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.Stage; + +public class MessageTelemetryTest { + @FunctionalInterface + private interface CheckedAction { void run() throws Throwable; } + + private static T assertThrows(Class type, CheckedAction action) { + try { action.run(); } + catch (Throwable failure) { + if (type.isInstance(failure)) return type.cast(failure); + throw new AssertionError("Expected " + type.getName(), failure); + } + throw new AssertionError("Expected " + type.getName()); + } + + private final List registrations = new ArrayList<>(); + private final ExecutorService worker = Executors.newSingleThreadExecutor(); + + private void install(Provider provider) { registrations.add(MessageTelemetry.install(provider)); } + + @Test public void destinationFinishClosesAfterFailedCallbackAndPreservesFirstFatalIdentity() throws Exception { + for (boolean same : List.of(false, true)) { + ThreadDeath failed = new ThreadDeath(); + Error closing = same ? failed : new OutOfMemoryError("close"); + List calls = new ArrayList<>(); + try (var registration = MessageTelemetry.install((stage, message) -> new Observation() { + public void failed(Throwable cause) { calls.add("failed"); throw failed; } + public void close() { calls.add("close"); throw closing; } + })) { + Observation observation = MessageTelemetry.start(Stage.DESTINATION, null); + assertSame(failed, assertThrows(ThreadDeath.class, + () -> MessageTelemetry.finish(observation, new IllegalStateException("engine")))); + assertEquals(List.of("failed", "close"), calls); + assertEquals(same ? 0 : 1, failed.getSuppressed().length); + if (!same) assertSame(closing, failed.getSuppressed()[0]); + } + } + } + + @Test public void destinationFinishCannotReplaceAnAlreadyCaughtEngineFatal() throws Exception { + ThreadDeath original = new ThreadDeath(), reporting = new ThreadDeath(), closing = new ThreadDeath(); + AtomicInteger closed = new AtomicInteger(); + try (var registration = MessageTelemetry.install((stage, message) -> new Observation() { + public void failed(Throwable failure) { assertSame(original, failure); throw reporting; } + public void close() { closed.incrementAndGet(); throw closing; } + })) { + MessageTelemetry.finish(MessageTelemetry.start(Stage.DESTINATION, null), original); + assertEquals(1, closed.get()); + assertArrayEquals(new Throwable[] { reporting, closing }, original.getSuppressed()); + } + MessageTelemetry.finish(null, original); + } + + @Test public void destinationFinishIsolatesOrdinaryCallbacksAndClosesOnce() throws Exception { + AtomicInteger closed = new AtomicInteger(); + try (var registration = MessageTelemetry.install((stage, message) -> new Observation() { + public void failed(Throwable failure) { throw new LinkageError("private callback detail"); } + public void close() { closed.incrementAndGet(); throw new AssertionError("private close detail"); } + })) { + MessageTelemetry.finish(MessageTelemetry.start(Stage.DESTINATION, null), new RuntimeException("engine")); + assertEquals(1, closed.get()); + } + } + + @After + public void cleanup() throws Exception { + Thread.interrupted(); + worker.shutdownNow(); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + for (AutoCloseable registration : registrations) registration.close(); + } + + @Test + public void absentProviderReturnsSameNoopAndOriginalTask() throws Exception { + Callable task = Object::new; + Observation noop = MessageTelemetry.start(Stage.SOURCE, null); + for (int i = 0; i < 1000; i++) { + assertSame(noop, MessageTelemetry.start(Stage.SEND, null)); + assertSame(task, MessageTelemetry.wrap(task)); + noop.status(Status.ERROR); + noop.failed(new IOException()); + noop.close(); + MessageTelemetry.beforeStore(null, null); + } + } + + @Test + public void registrationRejectsOverlapAndOldTokenCannotDetachReplacement() throws Exception { + AtomicInteger first = new AtomicInteger(), second = new AtomicInteger(); + Provider firstProvider = (stage, message) -> { first.incrementAndGet(); return null; }; + install(firstProvider); + assertThrows(IllegalStateException.class, () -> MessageTelemetry.install(firstProvider)); + MessageTelemetry.start(Stage.SOURCE, null).close(); + registrations.get(0).close(); + install((stage, message) -> { second.incrementAndGet(); return null; }); + registrations.get(0).close(); + MessageTelemetry.start(Stage.SOURCE, null).close(); + assertEquals(1, first.get()); + assertEquals(1, second.get()); + assertThrows(NullPointerException.class, () -> MessageTelemetry.install(null)); + } + + @Test + public void callbacksCannotReplaceBusinessFailureOrSkipClose() throws Exception { + AtomicInteger closed = new AtomicInteger(); + install((stage, message) -> new Observation() { + public void status(Status status) { throw new AssertionError("ordinary callback failure"); } + public void failed(Throwable cause) { throw new IllegalStateException("callback failure"); } + public void close() { closed.incrementAndGet(); throw new IllegalStateException("close failure"); } + }); + IOException original = new IOException("business failure"); + IOException thrown = assertThrows(IOException.class, () -> { + try (Observation observation = MessageTelemetry.start(Stage.SOURCE, null)) { + observation.status(Status.ERROR); + observation.failed(original); + throw original; + } + }); + assertSame(original, thrown); + assertEquals(0, thrown.getSuppressed().length); + assertEquals(1, closed.get()); + } + + @Test + public void startAndCaptureFailuresAreIsolated() throws Exception { + install(new Provider() { + public Observation start(Stage stage, ConnectorMessage message) { throw new LinkageError(); } + public Supplier capture() { throw new IllegalStateException(); } + }); + MessageTelemetry.start(Stage.SOURCE, null).close(); + Callable original = () -> "business result"; + assertSame(original, MessageTelemetry.wrap(original)); + assertEquals("business result", MessageTelemetry.wrap(original).call()); + } + + @Test + public void activationAndCleanupFailuresStillRunTaskExactlyOnce() throws Exception { + for (boolean failActivation : new boolean[] { true, false }) { + AtomicInteger executions = new AtomicInteger(), activations = new AtomicInteger(); + install(new Provider() { + public Observation start(Stage stage, ConnectorMessage message) { return null; } + public Supplier capture() { + return () -> { + activations.incrementAndGet(); + if (failActivation) throw new IllegalArgumentException(); + return () -> { throw new IllegalStateException(); }; + }; + } + }); + IOException original = new IOException("task failed"); + Callable wrapped = MessageTelemetry.wrap(() -> { executions.incrementAndGet(); throw original; }); + ExecutionException failure = assertThrows(ExecutionException.class, () -> worker.submit(wrapped).get(5, TimeUnit.SECONDS)); + assertSame(original, failure.getCause()); + assertEquals(1, activations.get()); + assertEquals(1, executions.get()); + registrations.get(registrations.size() - 1).close(); + } + } + + @Test + public void capturedContextRestoresWorkerOnSuccessFailureAndNestedExecutionAfterDetach() throws Exception { + ThreadLocal context = new ThreadLocal<>(); + AtomicInteger activeScopes = new AtomicInteger(); + install(contextProvider(context, activeScopes)); + worker.submit(() -> context.set("worker prior")).get(5, TimeUnit.SECONDS); + context.set("request"); + Callable success = MessageTelemetry.wrap(context::get); + IOException original = new IOException(); + Callable failure = MessageTelemetry.wrap(() -> { + assertEquals("request", context.get()); + context.set("nested"); + assertEquals("nested", MessageTelemetry.wrap(context::get).call()); + throw original; + }); + context.set("caller later"); + assertEquals("request", worker.submit(success).get(5, TimeUnit.SECONDS)); + assertSame(original, assertThrows(ExecutionException.class, () -> worker.submit(failure).get(5, TimeUnit.SECONDS)).getCause()); + assertEquals("worker prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + registrations.get(0).close(); + assertEquals("request", worker.submit(success).get(5, TimeUnit.SECONDS)); + assertEquals("worker prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + assertEquals("caller later", context.get()); + assertEquals(0, activeScopes.get()); + } + + @Test + public void rejectedAndCancelledBeforeStartTasksNeverActivate() throws Exception { + AtomicInteger scopes = new AtomicInteger(); + install(contextProvider(new ThreadLocal<>(), scopes)); + CountDownLatch blocked = new CountDownLatch(1), release = new CountDownLatch(1); + Future blocker = worker.submit(() -> { blocked.countDown(); release.await(); return null; }); + assertTrue(blocked.await(5, TimeUnit.SECONDS)); + AtomicInteger calls = new AtomicInteger(); + Future cancelled = worker.submit(MessageTelemetry.wrap(calls::incrementAndGet)); + assertTrue(cancelled.cancel(false)); + release.countDown(); + blocker.get(5, TimeUnit.SECONDS); + worker.submit(() -> {}).get(5, TimeUnit.SECONDS); + worker.shutdown(); + assertThrows(RejectedExecutionException.class, () -> worker.submit(MessageTelemetry.wrap(calls::incrementAndGet))); + assertEquals(0, calls.get()); + assertEquals(0, scopes.get()); + } + + @Test + public void cancellationAfterStartRestoresOnExecutingThread() throws Exception { + ThreadLocal context = new ThreadLocal<>(); + AtomicInteger scopes = new AtomicInteger(); + install(contextProvider(context, scopes)); + worker.submit(() -> context.set("prior")).get(5, TimeUnit.SECONDS); + CountDownLatch started = new CountDownLatch(1); + context.set("request"); + Future task = worker.submit(MessageTelemetry.wrap(() -> { + assertEquals("request", context.get()); + started.countDown(); + new CountDownLatch(1).await(); + return null; + })); + assertTrue(started.await(5, TimeUnit.SECONDS)); + assertTrue(task.cancel(true)); + assertEquals("prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + assertEquals(0, scopes.get()); + } + + @Test + public void fatalCallbacksRetainThrowableIdentity() throws Exception { + for (Error fatal : new Error[] { new OutOfMemoryError("synthetic"), new ThreadDeath() }) { + install((stage, message) -> { throw fatal; }); + assertSame(fatal, assertThrows(fatal.getClass(), () -> MessageTelemetry.start(Stage.SOURCE, null))); + registrations.get(registrations.size() - 1).close(); + } + } + + @Test + public void beforeStoreIsOptionalAndUsesTheCurrentExactInstallation() throws Exception { + Map source = new HashMap<>(); source.put("application", "preserved"); + install((stage,message) -> null); + MessageTelemetry.beforeStore(null, source); + assertEquals(1,source.size()); registrations.get(0).close(); + AtomicInteger calls = new AtomicInteger(); + install(new Provider() { + public Observation start(Stage stage, ConnectorMessage message) { return null; } + public void beforeStore(ConnectorMessage message, Map supplied) { + assertSame(source,supplied); calls.incrementAndGet(); supplied.put("private.context", "scalar"); + } + }); + registrations.get(0).close(); MessageTelemetry.beforeStore(null,source); + assertEquals(1,calls.get()); assertEquals("preserved",source.get("application")); assertEquals("scalar",source.get("private.context")); + registrations.get(1).close(); MessageTelemetry.beforeStore(null,source); assertEquals(1,calls.get()); + } + + @Test + public void beforeStoreIsolatesOrdinaryFailuresAndPreservesFatalIdentity() throws Exception { + for (Throwable failure : new Throwable[] {new IllegalStateException(), new LinkageError(), new AssertionError(), new OutOfMemoryError("synthetic"), new ThreadDeath()}) { + install(new Provider() { + public Observation start(Stage stage, ConnectorMessage message) { return null; } + public void beforeStore(ConnectorMessage message, Map source) { + if (failure instanceof Error) throw (Error)failure; + throw (RuntimeException)failure; + } + }); + if (failure instanceof VirtualMachineError || failure instanceof ThreadDeath) + assertSame(failure,assertThrows(failure.getClass(),()->MessageTelemetry.beforeStore(null,new HashMap<>()))); + else MessageTelemetry.beforeStore(null,new HashMap<>()); + registrations.get(registrations.size()-1).close(); + } + } + + private static Provider contextProvider(ThreadLocal context, AtomicInteger scopes) { + return new Provider() { + public Observation start(Stage stage, ConnectorMessage message) { return null; } + public Supplier capture() { + String captured = context.get(); + return () -> { + String prior = context.get(); + Thread owner = Thread.currentThread(); + context.set(captured); + scopes.incrementAndGet(); + return () -> { + assertSame(owner, Thread.currentThread()); + context.set(prior); + scopes.decrementAndGet(); + }; + }; + } + }; + } +} diff --git a/donkey/src/test/java/com/mirth/connect/donkey/test/DestinationScopeAdversarialTest.java b/donkey/src/test/java/com/mirth/connect/donkey/test/DestinationScopeAdversarialTest.java new file mode 100644 index 0000000000..9f10404343 --- /dev/null +++ b/donkey/src/test/java/com/mirth/connect/donkey/test/DestinationScopeAdversarialTest.java @@ -0,0 +1,159 @@ +/* Published under the Mozilla Public License 2.0. */ +package com.mirth.connect.donkey.test; + +import static org.junit.Assert.*; +import org.junit.*; +import java.lang.reflect.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.concurrent.locks.*; +import java.util.function.*; +import com.mirth.connect.donkey.model.message.*; +import com.mirth.connect.donkey.server.channel.*; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.Stage; +import com.mirth.connect.donkey.server.data.*; +import com.mirth.connect.donkey.test.MessageTelemetryHooksTest; +import com.mirth.connect.donkey.test.util.*; + +public class DestinationScopeAdversarialTest { + MessageTelemetryHooksTest fixture;TestChannel channel;Object recorder; + @BeforeClass public static void start()throws Exception{MessageTelemetryHooksTest.startEngine();} + @AfterClass public static void stop()throws Exception{MessageTelemetryHooksTest.stopEngine();} + @Before public void setup()throws Exception{fixture=new MessageTelemetryHooksTest();} + @After public void cleanup()throws Exception{if(fixture!=null)fixture.cleanup();} + static Object get(Object target,String name)throws Exception{Field field=target.getClass().getDeclaredField(name);field.setAccessible(true);return field.get(target);} + static void set(Object target,String name,Object value)throws Exception{Field field=target.getClass().getDeclaredField(name);field.setAccessible(true);field.set(target,value);} + static Object call(Object target,String name,Class[] types,Object...args)throws Exception{ + Method method=target.getClass().getDeclaredMethod(name,types);method.setAccessible(true); + try{return method.invoke(target,args);}catch(InvocationTargetException e){if(e.getCause() instanceof Error error)throw error;throw(Exception)e.getCause();} + } + void create(int destinations)throws Exception{call(fixture,"create",new Class[]{boolean.class,int.class},true,destinations);channel=(TestChannel)get(fixture,"channel");recorder=get(fixture,"recorder");} + void run()throws Exception{call(fixture,"runMessage",new Class[0]);} + TestDestinationConnector sender(Supplier send)throws Exception{return(TestDestinationConnector)call(fixture,"sender",new Class[]{Supplier.class},send);} + void queued(TestDestinationConnector destination,int retry)throws Exception{ + call(fixture,"queueProperties",new Class[]{TestDestinationConnector.class},destination); + ((TestConnectorProperties)destination.getConnectorProperties()).getDestinationConnectorProperties().setRetryIntervalMillis(retry); + } + ReentrantReadWriteLock queueLock(TestDestinationConnector destination)throws Exception{return(ReentrantReadWriteLock)get(destination.getQueue(),"statusUpdateLock");} + boolean queueThread(){return Thread.currentThread() instanceof DestinationConnector.DestinationQueueThread;} + Object current()throws Exception{return((ThreadLocal)get(recorder,"current")).get();} + boolean coarse()throws Exception{Object current=current();return current!=null&&get(current,"stage")==Stage.DESTINATION;} + List records(Stage stage,boolean queue)throws Exception{ + var selected=new ArrayList();for(Object r:(List)get(recorder,"records"))if(get(r,"stage")==stage&&(!queue||get(r,"thread") instanceof DestinationConnector.DestinationQueueThread))selected.add(r);return selected; + } + void after(Consumer action)throws Exception{set(recorder,"afterClose",action);} + static Object invoke(Object target,Method method,Object[] args)throws Throwable{try{return method.invoke(target,args);}catch(InvocationTargetException e){throw e.getCause();}} + void closeFailure(TestDestinationConnector destination,RuntimeException failure,boolean requireLock)throws Exception{ + DonkeyDaoFactory original=channel.getDaoFactory();AtomicBoolean armed=new AtomicBoolean(true); + channel.setDaoFactory((DonkeyDaoFactory)Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DonkeyDaoFactory.class},(proxy,method,args)->{ + Object value=invoke(original,method,args);if(!(value instanceof DonkeyDao dao))return value; + return Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DonkeyDao.class},(p,m,a)->{ + Object result=invoke(dao,m,a); + if(m.getName().equals("close")&&queueThread()&&coarse()&&(!requireLock||queueLock(destination).getReadHoldCount()>0)&&armed.compareAndSet(true,false))throw failure; + return result; + }); + })); + } + + @Test(timeout=15000) public void queueObservationClosesAfterFallbackStatusLockCleanupEvenWhenDaoCloseThrows()throws Exception{ + create(1);var destination=sender(()->new Response(Status.SENT,"reply"));queued(destination,1000); + RuntimeException original=new IllegalStateException("review DAO close");closeFailure(destination,original,true); + AtomicInteger heldAtClose=new AtomicInteger(-1);CountDownLatch closed=new CountDownLatch(1); + after(record->{try{if(get(record,"stage")==Stage.DESTINATION&&queueThread()){heldAtClose.set(queueLock(destination).getReadHoldCount());closed.countDown();}}catch(Exception e){throw new RuntimeException(e);}}); + run();assertTrue(closed.await(5,TimeUnit.SECONDS));destination.stopQueue(); + List attempts=records(Stage.DESTINATION,true);assertEquals(1,attempts.size());assertSame(original,get(attempts.get(0),"failure")); + assertEquals(0,queueLock(destination).getReadLockCount()); + System.out.println("STATUS_LOCK_AT_DESTINATION_CLOSE="+heldAtClose.get()); + assertEquals("coarse finish must follow the original fallback unlock",0,heldAtClose.get()); + } + + @Test(timeout=15000) public void caughtEngineFatalSurvivesLaterOrdinaryDaoCleanupAndCallbackFatal()throws Exception{ + create(1);ThreadDeath original=new ThreadDeath(),callback=new ThreadDeath();AtomicInteger sends=new AtomicInteger(); + var destination=sender(()->{if(sends.incrementAndGet()==1)throw original;return new Response(Status.SENT,"reply");});queued(destination,1000); + RuntimeException cleanup=new IllegalStateException("review later DAO close");closeFailure(destination,cleanup,false);CountDownLatch closed=new CountDownLatch(1); + after(record->{try{if(get(record,"stage")==Stage.DESTINATION&&queueThread()){closed.countDown();throw callback;}}catch(RuntimeException|Error e){throw e;}catch(Exception e){throw new RuntimeException(e);}}); + run();assertTrue(closed.await(5,TimeUnit.SECONDS));destination.stopQueue(); + assertSame(original,get(records(Stage.SEND,true).get(0),"failure")); + Object coarseFailure=get(records(Stage.DESTINATION,true).get(0),"failure"); + System.out.println("COARSE_FAILURE_AFTER_CAUGHT_FATAL="+coarseFailure.getClass().getName()); + assertSame("the caught engine fatal must remain the observation's primary failure",original,coarseFailure); + assertTrue(Arrays.asList(original.getSuppressed()).contains(callback)); + } + + @Test(timeout=15000) public void chainBodyFatalRemainsObservedWhenOrdinaryDaoCloseMasksIt() throws Exception { + chainFatalCleanup(new ThreadDeath(), new IllegalStateException("later chain close"), new ThreadDeath()); + } + + @Test(timeout=15000) public void chainFirstFatalRemainsObservedWhenDaoCloseAlsoThrowsFatal() throws Exception { + chainFatalCleanup(new OutOfMemoryError("original chain send"), new ThreadDeath(), new ThreadDeath()); + } + + private void chainFatalCleanup(Error original, Throwable cleanup, Error callback) throws Exception { + create(1); + AtomicBoolean failSend = new AtomicBoolean(); + sender(() -> { if (failSend.get()) throw original; return new Response(Status.SENT, "reply"); }); + run(); + ConnectorMessage message = (ConnectorMessage) get(records(Stage.DESTINATION, false).get(0), "message"); + var provider = channel.getDestinationChainProviders().get(0); + DonkeyDaoFactory originalFactory = channel.getDaoFactory(); + Method setDao = DestinationChainProvider.class.getDeclaredMethod("setDaoFactory", DonkeyDaoFactory.class); + setDao.setAccessible(true); + setDao.invoke(provider, Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{DonkeyDaoFactory.class}, (p, m, a) -> { + Object value = invoke(originalFactory, m, a); + if (!(value instanceof DonkeyDao dao)) return value; + return Proxy.newProxyInstance(getClass().getClassLoader(), new Class[]{DonkeyDao.class}, (proxy, method, args) -> { + Object result = invoke(dao, method, args); + if (method.getName().equals("close")) throw cleanup; + return result; + }); + })); + after(record -> { + try { if (get(record, "stage") == Stage.DESTINATION) throw callback; } + catch (RuntimeException | Error failure) { throw failure; } + catch (Exception failure) { throw new RuntimeException(failure); } + }); + failSend.set(true); + message.setStatus(Status.RECEIVED); + try { + var chain = new DestinationChain(provider); chain.setMessage(message); + try { chain.call(); fail("DAO close failure was lost"); } + catch (RuntimeException | Error actual) { assertSame("original engine finally behavior", cleanup, actual); } + assertSame(original, get(records(Stage.SEND, false).get(1), "failure")); + Object destination = records(Stage.DESTINATION, false).get(1); + assertSame(original, get(destination, "failure")); + assertEquals(1, get(destination, "closed")); + assertTrue(Arrays.asList(original.getSuppressed()).contains(callback)); + assertNull(current()); + } finally { + setDao.invoke(provider, originalFactory); + message.setStatus(Status.SENT); + } + } + + @Test(timeout=15000) public void unavailableQueueDaoIsObservedAndActualRetryUsesFreshScope()throws Exception{ + create(1);AtomicInteger sends=new AtomicInteger();var destination=sender(()->{sends.incrementAndGet();return new Response(Status.SENT,"reply");});queued(destination,1); + DonkeyDaoFactory original=channel.getDaoFactory();AtomicBoolean armed=new AtomicBoolean(true);RuntimeException unavailable=new IllegalStateException("review unavailable queue DAO"); + channel.setDaoFactory((DonkeyDaoFactory)Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DonkeyDaoFactory.class},(proxy,method,args)->{ + if(method.getName().equals("getDao")&&queueThread()&&coarse()&&armed.compareAndSet(true,false))throw unavailable; + return invoke(original,method,args); + })); + run();call(fixture,"awaitQueuedCompletion",new Class[0]); + List attempts=records(Stage.DESTINATION,true);assertEquals(2,attempts.size());assertSame(unavailable,get(attempts.get(0),"failure")); + assertEquals(Status.QUEUED,get(attempts.get(0),"status"));assertEquals(Status.SENT,get(attempts.get(1),"status"));assertEquals(1,sends.get());assertNotSame(attempts.get(0),attempts.get(1)); + for(Object r:attempts)assertEquals(1,get(r,"closed")); + } + + @Test(timeout=15000) public void sequentialAndParallelDestinationsOwnIndependentCoarseScopes()throws Exception{ + create(3);var providers=channel.getDestinationChainProviders();var first=providers.get(0);var second=providers.get(1); + first.addDestination(2,second.getDestinationConnectors().get(2));providers.remove(1); + AtomicBoolean previousClosed=new AtomicBoolean(); + set(recorder,"beforeStage",(BiConsumer)(stage,message)->{if(stage==Stage.DESTINATION&&message.getMetaDataId()==2)try{ + previousClosed.set(records(Stage.DESTINATION,false).stream().filter(r->{try{return((ConnectorMessage)get(r,"message")).getMetaDataId()==1;}catch(Exception e){throw new RuntimeException(e);}}).allMatch(r->{try{return(Integer)get(r,"closed")==1;}catch(Exception e){throw new RuntimeException(e);}})); + }catch(Exception e){throw new RuntimeException(e);}}); + run();assertTrue(previousClosed.get());assertEquals(3,records(Stage.DESTINATION,false).size());assertEquals(3,records(Stage.SEND,false).size()); + Object source=records(Stage.SOURCE,false).get(0);long id=((ConnectorMessage)get(source,"message")).getMessageId(); + for(Object coarse:records(Stage.DESTINATION,false)){assertSame(source,get(coarse,"parent"));assertEquals(1,get(coarse,"closed"));ConnectorMessage msg=(ConnectorMessage)get(coarse,"message");TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),id,msg.getMetaDataId(),Status.SENT);} + for(Object send:records(Stage.SEND,false)){Object parent=get(send,"parent");assertEquals(Stage.DESTINATION,get(parent,"stage"));assertSame(get(send,"message"),get(parent,"message"));} + } +} diff --git a/donkey/src/test/java/com/mirth/connect/donkey/test/DestinationScopeParityTest.java b/donkey/src/test/java/com/mirth/connect/donkey/test/DestinationScopeParityTest.java new file mode 100644 index 0000000000..883ba8c9e1 --- /dev/null +++ b/donkey/src/test/java/com/mirth/connect/donkey/test/DestinationScopeParityTest.java @@ -0,0 +1,94 @@ +/* Published under the Mozilla Public License 2.0. */ +package com.mirth.connect.donkey.test; + +import static org.junit.Assert.*; +import org.junit.*; +import java.lang.reflect.*; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.concurrent.locks.*; +import java.util.function.*; +import com.mirth.connect.donkey.model.message.*; +import com.mirth.connect.donkey.server.channel.*; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.Stage; +import com.mirth.connect.donkey.server.data.*; +import com.mirth.connect.donkey.test.MessageTelemetryHooksTest; +import com.mirth.connect.donkey.test.util.*; + +public class DestinationScopeParityTest { + MessageTelemetryHooksTest fixture;TestChannel channel;Object recorder; + @BeforeClass public static void start()throws Exception{MessageTelemetryHooksTest.startEngine();} + @AfterClass public static void stop()throws Exception{MessageTelemetryHooksTest.stopEngine();} + @Before public void setup()throws Exception{fixture=new MessageTelemetryHooksTest();} + @After public void cleanup()throws Exception{if(fixture!=null)fixture.cleanup();} + static Object get(Object target,String name)throws Exception{Field field=target.getClass().getDeclaredField(name);field.setAccessible(true);return field.get(target);} + static void set(Object target,String name,Object value)throws Exception{Field field=target.getClass().getDeclaredField(name);field.setAccessible(true);field.set(target,value);} + static Object call(Object target,String name,Class[] types,Object...args)throws Exception{ + Method method=target.getClass().getDeclaredMethod(name,types);method.setAccessible(true); + try{return method.invoke(target,args);}catch(InvocationTargetException e){if(e.getCause() instanceof Error error)throw error;throw(Exception)e.getCause();} + } + void create(int destinations)throws Exception{call(fixture,"create",new Class[]{boolean.class,int.class},true,destinations);channel=(TestChannel)get(fixture,"channel");recorder=get(fixture,"recorder");} + void run()throws Exception{call(fixture,"runMessage",new Class[0]);} + TestDestinationConnector sender(Supplier send)throws Exception{return(TestDestinationConnector)call(fixture,"sender",new Class[]{Supplier.class},send);} + void queued(TestDestinationConnector destination,int retry)throws Exception{ + call(fixture,"queueProperties",new Class[]{TestDestinationConnector.class},destination); + ((TestConnectorProperties)destination.getConnectorProperties()).getDestinationConnectorProperties().setRetryIntervalMillis(retry); + } + ReentrantReadWriteLock queueLock(TestDestinationConnector destination)throws Exception{return(ReentrantReadWriteLock)get(destination.getQueue(),"statusUpdateLock");} + boolean queueThread(){return Thread.currentThread() instanceof DestinationConnector.DestinationQueueThread;} + Object current()throws Exception{return((ThreadLocal)get(recorder,"current")).get();} + boolean coarse()throws Exception{Object current=current();return current!=null&&get(current,"stage")==Stage.DESTINATION;} + List records(Stage stage,boolean queue)throws Exception{ + var selected=new ArrayList();for(Object r:(List)get(recorder,"records"))if(get(r,"stage")==stage&&(!queue||get(r,"thread") instanceof DestinationConnector.DestinationQueueThread))selected.add(r);return selected; + } + void after(Consumer action)throws Exception{set(recorder,"afterClose",action);} + static Object invoke(Object target,Method method,Object[] args)throws Throwable{try{return method.invoke(target,args);}catch(InvocationTargetException e){throw e.getCause();}} + void closeFailure(TestDestinationConnector destination,RuntimeException failure,boolean requireLock)throws Exception{ + DonkeyDaoFactory original=channel.getDaoFactory();AtomicBoolean armed=new AtomicBoolean(true); + channel.setDaoFactory((DonkeyDaoFactory)Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DonkeyDaoFactory.class},(proxy,method,args)->{ + Object value=invoke(original,method,args);if(!(value instanceof DonkeyDao dao))return value; + return Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DonkeyDao.class},(p,m,a)->{ + Object result=invoke(dao,m,a); + if(m.getName().equals("close")&&queueThread()&&coarse()&&(!requireLock||queueLock(destination).getReadHoldCount()>0)&&armed.compareAndSet(true,false))throw failure; + return result; + }); + })); + } + + + @Test(timeout=15000) public void rotatedQueueKeepsMessageOrderAndStartsFreshCoarseScope()throws Exception{ + create(1);CountDownLatch entered=new CountDownLatch(1),release=new CountDownLatch(1);AtomicInteger calls=new AtomicInteger();List order=new CopyOnWriteArrayList<>(); + var destination=sender(()->{try{order.add(((ConnectorMessage)get(current(),"message")).getMessageId());if(calls.incrementAndGet()==1){entered.countDown();if(!release.await(5,TimeUnit.SECONDS))throw new AssertionError("first send not released");return new Response(Status.QUEUED,"retry");}return new Response(Status.SENT,"reply");}catch(Exception e){throw new RuntimeException(e);}}); + queued(destination,1);((TestConnectorProperties)destination.getConnectorProperties()).getDestinationConnectorProperties().setRotate(true); + try{run();assertTrue(entered.await(5,TimeUnit.SECONDS));((TestSourceConnector)channel.getSourceConnector()).readTestMessage("second");}finally{release.countDown();} + long end=System.nanoTime()+TimeUnit.SECONDS.toNanos(5); + while(true){try{assertEquals(3,calls.get());for(Object source:records(Stage.SOURCE,false)){long id=((ConnectorMessage)get(source,"message")).getMessageId();TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),id,1,Status.SENT);assertFalse(destination.getQueue().isCheckedOut(id));}for(Object coarse:records(Stage.DESTINATION,true))assertEquals(1,get(coarse,"closed"));break;}catch(AssertionError waiting){if(System.nanoTime()>end)throw waiting;Thread.sleep(5);}} + destination.stopQueue();List sources=records(Stage.SOURCE,false);assertEquals(2,sources.size());long first=((ConnectorMessage)get(sources.get(0),"message")).getMessageId(),second=((ConnectorMessage)get(sources.get(1),"message")).getMessageId(); + assertEquals(Arrays.asList(first,second,first),order);List attempts=records(Stage.DESTINATION,true);assertEquals(3,attempts.size());assertNotSame(attempts.get(0),attempts.get(2)); + List sends=records(Stage.SEND,true);for(int i=0;i<3;i++){assertSame(attempts.get(i),get(sends.get(i),"parent"));assertNull(get(attempts.get(i),"parent"));} + } + + static final class FailingCloneProperties extends TestConnectorProperties { + transient AtomicInteger clones; transient RuntimeException failure; + FailingCloneProperties(AtomicInteger clones,RuntimeException failure){this.clones=clones;this.failure=failure;} + @Override public com.mirth.connect.donkey.model.channel.ConnectorProperties clone(){if(Thread.currentThread() instanceof DestinationConnector.DestinationQueueThread&&clones.incrementAndGet()==1)throw failure;return this;} + } + @Test(timeout=15000) public void queuePropertyCloneFailureClosesAndReleasesForRetry()throws Exception{ + create(1);AtomicInteger sends=new AtomicInteger(),clones=new AtomicInteger();RuntimeException original=new IllegalStateException("review properties"); + var destination=sender(()->{sends.incrementAndGet();return new Response(Status.SENT,"reply");}); + destination.setConnectorProperties(new FailingCloneProperties(clones,original));queued(destination,1); + run();call(fixture,"awaitQueuedCompletion",new Class[0]);List attempts=records(Stage.DESTINATION,true);assertEquals(2,attempts.size());assertSame(original,get(attempts.get(0),"failure"));assertEquals(Status.QUEUED,get(attempts.get(0),"status"));assertEquals(Status.SENT,get(attempts.get(1),"status"));assertEquals(1,sends.get()); + } + + @Test(timeout=15000) public void nullResponsePreservesActualEngineErrorAndHasNoResponseStage()throws Exception{ + create(1);AtomicInteger sends=new AtomicInteger();sender(()->{sends.incrementAndGet();return null;});run();Object send=records(Stage.SEND,false).get(0),coarse=records(Stage.DESTINATION,false).get(0); + assertEquals(1,sends.get());assertTrue(get(send,"failure") instanceof NullPointerException);assertSame(get(send,"failure"),get(coarse,"failure"));assertEquals(Status.ERROR,get(coarse,"status"));assertEquals(0,records(Stage.RESPONSE,false).size());ConnectorMessage msg=(ConnectorMessage)get(coarse,"message");TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),msg.getMessageId(),1,Status.ERROR); + } + + @Test(timeout=15000) public void missingChainDaoDoesNotCreatePartialCoarseScopeOrRetryBusinessTask()throws Exception{ + create(1);run();Object first=records(Stage.DESTINATION,false).get(0);ConnectorMessage msg=(ConnectorMessage)get(first,"message");msg.setStatus(Status.PENDING);int before=records(Stage.DESTINATION,false).size();var provider=channel.getDestinationChainProviders().get(0);DonkeyDaoFactory original=channel.getDaoFactory();RuntimeException unavailable=new IllegalStateException("review missing chain DAO"); + Method setDao=DestinationChainProvider.class.getDeclaredMethod("setDaoFactory",DonkeyDaoFactory.class);setDao.setAccessible(true);setDao.invoke(provider,Proxy.newProxyInstance(getClass().getClassLoader(),new Class[]{DonkeyDaoFactory.class},(p,m,a)->{if(m.getName().equals("getDao"))throw unavailable;return invoke(original,m,a);})); + try{var chain=new DestinationChain(provider);chain.setMessage(msg);try{chain.call();fail("missing DAO accepted");}catch(RuntimeException actual){assertSame(unavailable,actual);}assertEquals(before,records(Stage.DESTINATION,false).size());assertNull(current());}finally{setDao.invoke(provider,original);msg.setStatus(Status.SENT);} + } +} diff --git a/donkey/src/test/java/com/mirth/connect/donkey/test/MessageTelemetryControllerIsolationTest.java b/donkey/src/test/java/com/mirth/connect/donkey/test/MessageTelemetryControllerIsolationTest.java new file mode 100644 index 0000000000..22877a1962 --- /dev/null +++ b/donkey/src/test/java/com/mirth/connect/donkey/test/MessageTelemetryControllerIsolationTest.java @@ -0,0 +1,47 @@ +/* + * The software in this package is published under the terms of the MPL license a copy of which + * has been included with this distribution in the LICENSE.txt file. + */ +package com.mirth.connect.donkey.test; + +import static org.junit.Assert.*; +import org.junit.*; +import java.lang.reflect.Field; +import java.util.Collections; +import com.mirth.connect.donkey.server.Donkey; +import com.mirth.connect.donkey.server.DonkeyConnectionPools; +import com.mirth.connect.donkey.server.controllers.ChannelController; +import com.mirth.connect.donkey.server.controllers.MessageController; +import com.mirth.connect.donkey.test.MessageTelemetryHooksTest; + +public class MessageTelemetryControllerIsolationTest { + private final Class[] types={Donkey.class,DonkeyConnectionPools.class,ChannelController.class,MessageController.class}; + private Object[] prior; + private static Field singleton(Class type)throws Exception{Field field=type.getDeclaredField("instance");field.setAccessible(true);return field;} + @Before public void isolateProbe()throws Exception{ + prior=new Object[types.length]; + for(int i=0;i[] singletons={Donkey.class,DonkeyConnectionPools.class,ChannelController.class}; + private static Field singleton(Class type)throws Exception{Field field=type.getDeclaredField("instance");field.setAccessible(true);return field;} + @Before public void isolateProbe()throws Exception{ + originals=new Object[singletons.length]; + for(int i=0;i r.stage == Stage.SOURCE && r.closed == 1)) { + if (System.nanoTime() >= deadline) throw new AssertionError("source did not finish"); + Thread.sleep(5); + } + } + + @Test + public void synchronousScopesCoverActualTransformsSendResponseAndRestoreCaller() throws Exception { + create(true, 1); + runMessage(); + assertHierarchy(1); + Recorded source = recorder.only(Stage.SOURCE, 0); + assertSame(Thread.currentThread(), source.thread); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), source.message.getMessageId(), 0, Status.TRANSFORMED); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), source.message.getMessageId(), 1, Status.SENT); + assertEquals(Status.SENT, recorder.only(Stage.SEND, 1).status); + assertEquals(Status.SENT, recorder.only(Stage.RESPONSE, 1).status); + } + + @Test + public void parallelDestinationWorkerSeesSourceAndHasIndependentMessageMap() throws Exception { + create(true, 2); + runMessage(); + assertHierarchy(2); + Recorded source = recorder.only(Stage.SOURCE, 0); + assertNotSame(source.thread, recorder.only(Stage.SEND, 1).thread); + assertSame(source.thread, recorder.only(Stage.SEND, 2).thread); + assertNotSame(recorder.only(Stage.SEND, 1).message.getChannelMap(), recorder.only(Stage.SEND, 2).message.getChannelMap()); + } + + @Test + public void queuedSourceStartsScopeOnWorkerAndCompletesStoredMessage() throws Exception { + create(false, 2); + runMessage(); + assertHierarchy(2); + Recorded source = recorder.only(Stage.SOURCE, 0); + assertNotSame(Thread.currentThread(), source.thread); + for (int id = 1; id <= 2; id++) + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), source.message.getMessageId(), id, Status.SENT); + } + + @Test + public void preparationPersistsOnIngressBeforeQueuedSourceProcessingAndFreezesSourceKeys() throws Exception { + preparationPersistsBeforeProcessing(true); + } + + @Test + public void preparationPersistsWithOnlyInitialRawDurabilityAndNoLaterMapStorage() throws Exception { + preparationPersistsBeforeProcessing(false); + } + + private void preparationPersistsBeforeProcessing(boolean storeMaps) throws Exception { + create(false,1); + channel.getStorageSettings().setStoreMaps(storeMaps); + channel.getStorageSettings().setRawDurable(true); + java.util.concurrent.atomic.AtomicReference prepared = new java.util.concurrent.atomic.AtomicReference<>(); + java.util.concurrent.atomic.AtomicReference ingress = new java.util.concurrent.atomic.AtomicReference<>(); + CountDownLatch processing = new CountDownLatch(1), release = new CountDownLatch(1); + Map carrier = new HashMap<>(); carrier.put("traceparent","content-free-fixture"); + recorder.preparation = (message,map) -> { assertNull(prepared.getAndSet(message)); ingress.set(Thread.currentThread()); map.put("oie.test.context",carrier); }; + recorder.sourceStarted = () -> { + processing.countDown(); + try { assertTrue(release.await(5,TimeUnit.SECONDS)); } + catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw new IllegalStateException(interrupted); } + }; + channel.deploy(); channel.start(null); + try { + ((TestSourceConnector)channel.getSourceConnector()).readTestMessage("application content"); + assertTrue(processing.await(5,TimeUnit.SECONDS)); + ConnectorMessage message = prepared.get(); assertNotNull(message); assertSame(Thread.currentThread(),ingress.get()); + assertTrue(recorder.records.isEmpty()); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),message.getMessageId(),0,Status.RECEIVED); + var content = TestUtils.getMessageContent(channel.getChannelId(),message.getMessageId(),0, + com.mirth.connect.donkey.model.message.ContentType.SOURCE_MAP); + Map stored = Donkey.getInstance().getSerializer().deserialize(content.getContent(),Map.class); + assertEquals(carrier,stored.get("oie.test.context")); + try { message.getSourceMap().put("application-overwrite","forbidden"); fail("source keys must be read-only"); } + catch (UnsupportedOperationException expected) { } + } finally { release.countDown(); } + long until=System.nanoTime()+TimeUnit.SECONDS.toNanos(5); + while (recorder.records.stream().noneMatch(r->r.stage==Stage.SOURCE && r.closed==1)) { + assertTrue(System.nanoTime()-until<0);Thread.sleep(2); + } + assertHierarchy(1); + assertEquals(carrier,recorder.only(Stage.SEND,1).message.getSourceMap().get("oie.test.context")); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),prepared.get().getMessageId(),1,Status.SENT); + } + + @Test + public void failedPreparationDoesNotChangeActualMessageCompletion() throws Exception { + create(true,1); recorder.preparation=(message,map)->{throw new LinkageError("private callback detail");}; + runMessage(); assertHierarchy(1); + Recorded source=recorder.only(Stage.SOURCE,0); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),source.message.getMessageId(),1,Status.SENT); + } + + @Test + public void fatalPreparationUsesExistingDispatchErrorAndReleasesTransactionAndProcessLock() throws Exception { + create(true,1); + Error original = new ThreadDeath(); + java.util.concurrent.atomic.AtomicReference prepared = new java.util.concurrent.atomic.AtomicReference<>(); + recorder.preparation=(message,map)->{ prepared.set(message); throw original; }; + channel.deploy(); channel.start(null); String threadName=Thread.currentThread().getName(); + try { ((TestSourceConnector)channel.getSourceConnector()).readTestMessage("before failure"); fail("dispatch must fail"); } + catch (com.mirth.connect.donkey.server.channel.ChannelException expected) { assertSame(original,expected.getCause()); } + assertEquals(threadName,Thread.currentThread().getName()); assertTrue(recorder.records.isEmpty()); assertNotNull(prepared.get()); + com.mirth.connect.donkey.model.message.Message absent = new com.mirth.connect.donkey.model.message.Message(); + absent.setChannelId(channel.getChannelId()); absent.setMessageId(prepared.get().getMessageId()); + TestUtils.assertMessageDoesNotExist(absent); + recorder.preparation=(message,map)->{}; + ((TestSourceConnector)channel.getSourceConnector()).readTestMessage("after failure"); + assertHierarchy(1); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(),recorder.only(Stage.SOURCE,0).message.getMessageId(),1,Status.SENT); + } + + @Test + public void filteredSourceClosesEarlyWithoutStartingDestinations() throws Exception { + create(true, 1); + ((TestFilterTransformer) channel.getSourceConnector().getFilterTransformerExecutor().getFilterTransformer()).setFiltered(true); + runMessage(); + assertEquals(2, recorder.records.size()); + Recorded source = recorder.only(Stage.SOURCE, 0); + assertEquals(Status.FILTERED, source.status); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), source.message.getMessageId(), 0, Status.FILTERED); + } + + @Test + public void originalTransformFailureIsObservedAndEngineStillStoresError() throws Exception { + create(true, 1); + FilterTransformerException original = new FilterTransformerException("deliberate transform failure", null, "test"); + channel.getSourceConnector().getFilterTransformerExecutor().setFilterTransformer(new TestFilterTransformer() { + @Override public FilterTransformerResult doFilterTransform(ConnectorMessage message) throws FilterTransformerException { throw original; } + }); + runMessage(); + assertEquals(2, recorder.records.size()); + assertSame(original, recorder.only(Stage.TRANSFORM, 0).failure); + Recorded source = recorder.only(Stage.SOURCE, 0); + assertEquals(Status.ERROR, source.status); + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), source.message.getMessageId(), 0, Status.ERROR); + } + + @Test + public void brokenCallbacksLeaveRealChannelAndTransactionOutcomeIntact() throws Exception { + create(true, 2); + recorder.failCallbacks = true; + runMessage(); + assertHierarchy(2); + Recorded source = recorder.only(Stage.SOURCE, 0); + for (int id = 1; id <= 2; id++) + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), source.message.getMessageId(), id, Status.SENT); + } + + private List records(Stage stage, int connector) { + List result = new ArrayList<>(); + for (Recorded record : recorder.records) + if (record.stage == stage && record.message.getMetaDataId() == connector) result.add(record); + return result; + } + private Recorded only(Stage stage, int connector) { return recorder.only(stage, connector); } + private void status(int connector, Status status) throws Exception { + TestUtils.assertConnectorMessageStatusEquals(channel.getChannelId(), only(Stage.SOURCE, 0).message.getMessageId(), connector, status); + } + private TestDestinationConnector sender(Supplier body) throws Exception { + return sender(body, message -> { }); + } + private TestDestinationConnector sender(Supplier body, java.util.function.Consumer replacement) throws Exception { + TestDestinationConnector destination = new TestDestinationConnector() { + @Override public Response send(ConnectorProperties properties, ConnectorMessage message) { return body.get(); } + @Override public void replaceConnectorProperties(ConnectorProperties properties, ConnectorMessage message) { replacement.accept(message); } + }; + destination.setChannel(channel); + TestUtils.initDestinationConnector(destination, channel.getChannelId(), channel.getServerId(), new TestConnectorProperties(), "review destination", new TestDataType(), new TestDataType(), new TestResponseTransformer(), 1); + destination.setMetaDataReplacer(channel.getSourceConnector().getMetaDataReplacer()); + destination.setMetaDataColumns(channel.getMetaDataColumns()); + destination.setFilterTransformerExecutor(TestUtils.createDefaultFilterTransformerExecutor()); + channel.getDestinationChainProviders().get(0).addDestination(1, destination); + return destination; + } + + private void awaitQueuedCompletion() throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (true) { + try { + status(1, Status.SENT); + assertFalse(channel.getDestinationConnector(1).getQueue().isCheckedOut(only(Stage.SOURCE, 0).message.getMessageId())); + assertTrue(recorder.records.stream().filter(r -> r.stage == Stage.DESTINATION).allMatch(r -> r.closed == 1)); + return; + } catch (AssertionError waiting) { + if (System.nanoTime() - deadline >= 0) throw waiting; + Thread.sleep(5); + } + } + } + + private DestinationConnectorProperties queueProperties(TestDestinationConnector destination) { + var props = ((TestConnectorProperties) destination.getConnectorProperties()).getDestinationConnectorProperties(); + props.setQueueEnabled(true); props.setSendFirst(false); props.setRegenerateTemplate(true); props.setRetryIntervalMillis(1); + return props; + } + + @Test public void sentTraversalSkipsDestinationObservation() throws Exception { + create(true, 1); runMessage(); + ConnectorMessage sent = only(Stage.SEND, 1).message; + assertEquals(Status.SENT, sent.getStatus()); + int before = recorder.records.size(); + var chain = new com.mirth.connect.donkey.server.channel.DestinationChain(channel.getDestinationChainProviders().get(0)); + chain.setMessage(sent); assertEquals(1, chain.call().size()); + assertEquals(before, recorder.records.size()); + status(1, Status.SENT); + } + + @Test public void actualPersistedPendingDestinationCreatesOnlyCoarseAndResponseScopes() throws Exception { + create(true, 1); runMessage(); + ConnectorMessage sent = only(Stage.SEND, 1).message; + var dao = channel.getDaoFactory().getDao(); + try { sent.setStatus(Status.PENDING); dao.updateStatus(sent, Status.SENT); dao.commit(true); } + finally { dao.close(); } + ConnectorMessage restored; + dao = channel.getDaoFactory().getDao(); + try { restored = dao.getConnectorMessages(channel.getChannelId(), sent.getMessageId(), java.util.Set.of(1), true).get(0); } + finally { dao.close(); } + assertNotSame(sent, restored); assertEquals(Status.PENDING, restored.getStatus()); + int before = recorder.records.size(); + var chain = new com.mirth.connect.donkey.server.channel.DestinationChain(channel.getDestinationChainProviders().get(0)); + chain.setMessage(restored); chain.call(); + assertEquals(before + 2, recorder.records.size()); + Recorded coarse = records(Stage.DESTINATION, 1).get(1), response = records(Stage.RESPONSE, 1).get(1); + assertNull(coarse.parent); assertSame(coarse, response.parent); + assertEquals(1, records(Stage.SEND, 1).size()); assertEquals(1, records(Stage.SOURCE, 0).size()); + assertEquals(Status.SENT, coarse.status); status(1, Status.SENT); + } + + @Test public void fatalQueueStartReleasesAcquiredMessageForRealRetry() throws Exception { + create(true, 1); + AtomicInteger starts = new AtomicInteger(), sends = new AtomicInteger(); + var destination = sender(() -> { sends.incrementAndGet(); return new Response(Status.SENT, "reply"); }); + queueProperties(destination); + recorder.beforeStage = (stage, message) -> { + if (stage == Stage.DESTINATION && Thread.currentThread() instanceof DestinationConnector.DestinationQueueThread + && starts.incrementAndGet() == 1) throw new ThreadDeath(); + }; + runMessage(); awaitQueuedCompletion(); + assertEquals(2, starts.get()); assertEquals(1, sends.get()); + assertEquals(2, records(Stage.DESTINATION, 1).size()); + } + + @Test public void queuePreSendFailureIsObservedEvenWhileStatusRemainsQueued() throws Exception { + create(true, 1); + AtomicInteger replacements = new AtomicInteger(), sends = new AtomicInteger(); + RuntimeException original = new IllegalStateException("private pre-send failure"); + var destination = sender(() -> { sends.incrementAndGet(); return new Response(Status.SENT, "reply"); }, message -> { + if (Thread.currentThread() instanceof DestinationConnector.DestinationQueueThread + && replacements.incrementAndGet() == 1) throw original; + }); + queueProperties(destination); + runMessage(); awaitQueuedCompletion(); + List attempts = records(Stage.DESTINATION, 1).stream() + .filter(r -> r.thread instanceof DestinationConnector.DestinationQueueThread).toList(); + assertEquals(2, attempts.size()); assertSame(original, attempts.get(0).failure); + assertEquals(Status.QUEUED, attempts.get(0).status); assertEquals(Status.SENT, attempts.get(1).status); + assertEquals(1, sends.get()); assertSame(attempts.get(1), only(Stage.SEND, 1).parent); + } + + @Test public void fatalQueueCloseRunsAfterEntryAndStatusLockRelease() throws Exception { + create(true, 1); + var destination = sender(() -> new Response(Status.SENT, "reply")); + queueProperties(destination); + AtomicInteger closes = new AtomicInteger(); + recorder.afterClose = record -> { + if (record.stage == Stage.DESTINATION && record.thread instanceof DestinationConnector.DestinationQueueThread) { + try { + assertFalse(destination.getQueue().isCheckedOut(record.message.getMessageId())); + Field field = destination.getQueue().getClass().getDeclaredField("statusUpdateLock"); field.setAccessible(true); + assertEquals(0, ((java.util.concurrent.locks.ReentrantReadWriteLock) field.get(destination.getQueue())).getReadLockCount()); + closes.incrementAndGet(); + } catch (ReflectiveOperationException failure) { throw new AssertionError(failure); } + throw new ThreadDeath(); + } + }; + runMessage(); awaitQueuedCompletion(); assertEquals(1, closes.get()); + assertEquals(Status.SENT, only(Stage.SEND, 1).status); + } + @Test public void checkedPreprocessorFailureStoresSourceErrorWithoutDestinationScopes() throws Exception { + create(true, 1); + DonkeyException original = new DonkeyException("preprocessor", null, "formatted"); + channel.setPreProcessor(message -> { throw original; }); runMessage(); + status(0, Status.ERROR); assertEquals(Status.ERROR, only(Stage.SOURCE,0).status); + assertTrue(records(Stage.TRANSFORM,0).isEmpty()); assertTrue(records(Stage.SEND,1).isEmpty()); + } + @Test public void filteredDestinationKeepsSourceTransformedAndSkipsSend() throws Exception { + create(true, 1); + ((TestFilterTransformer) channel.getDestinationConnector(1).getFilterTransformerExecutor().getFilterTransformer()).setFiltered(true); + runMessage(); status(0,Status.TRANSFORMED); status(1,Status.FILTERED); + assertEquals(Status.FILTERED, only(Stage.TRANSFORM,1).status); + assertTrue(records(Stage.SEND,1).isEmpty()); assertTrue(records(Stage.RESPONSE,1).isEmpty()); + } + @Test public void checkedDestinationTransformFailurePreservesCauseAndStoresError() throws Exception { + create(true, 1); + FilterTransformerException original = new FilterTransformerException("transform", null, "formatted"); + channel.getDestinationConnector(1).getFilterTransformerExecutor().setFilterTransformer(new TestFilterTransformer() { + public FilterTransformerResult doFilterTransform(ConnectorMessage message) throws FilterTransformerException { throw original; } + }); + runMessage(); status(0, Status.TRANSFORMED); status(1, Status.ERROR); + assertSame(original, only(Stage.TRANSFORM,1).failure); assertTrue(records(Stage.SEND,1).isEmpty()); + } + @Test public void rawSendStatusIsVisibleBeforeQueueRulesAlterIt() throws Exception { + create(true, 1); + sender(() -> new Response(Status.QUEUED,"reply")); runMessage(); + assertEquals(Status.QUEUED,only(Stage.SEND,1).status); + assertEquals(Status.ERROR,only(Stage.RESPONSE,1).status); + status(0,Status.TRANSFORMED); status(1,Status.ERROR); + } + @Test public void synchronousRetryProducesTwoSendsAndOneFinalResponse() throws Exception { + create(true, 1); + AtomicInteger calls = new AtomicInteger(); + TestDestinationConnector d = sender(() -> new Response(calls.incrementAndGet() == 1 ? Status.ERROR : Status.SENT,"reply")); + DestinationConnectorProperties props = ((TestConnectorProperties)d.getConnectorProperties()).getDestinationConnectorProperties(); + props.setRetryCount(1); props.setRetryIntervalMillis(1); runMessage(); + assertEquals(2,calls.get()); List sends=records(Stage.SEND,1); assertEquals(2,sends.size()); + assertEquals(Status.ERROR,sends.get(0).status); assertEquals(Status.SENT,sends.get(1).status); + assertEquals(Status.SENT,only(Stage.RESPONSE,1).status); status(1,Status.SENT); + assertEquals(2,((ConnectorMessage)sends.get(1).message).getSendAttempts()); + } + @Test public void checkedResponseTransformFailurePreservesCauseAndStoresError() throws Exception { + create(true, 1); + ResponseTransformerException original = new ResponseTransformerException("response",null,"formatted"); + channel.getDestinationConnector(1).getResponseTransformerExecutor().setResponseTransformer(new TestResponseTransformer() { + public String doTransform(Response response, ConnectorMessage message) throws DonkeyException { throw original; } + }); + runMessage(); status(0,Status.TRANSFORMED); status(1,Status.ERROR); + assertSame(original,only(Stage.RESPONSE,1).failure); + assertEquals(Status.SENT,only(Stage.SEND,1).status); + } + @Test public void validationExceptionIsSendFailureAndDoesNotRunResponseTransformer() throws Exception { + create(true, 1); + RuntimeException original = new IllegalStateException("validator failed"); + TestDestinationConnector d=sender(() -> new Response(Status.SENT,"reply","","",true)); + d.setResponseValidator((response,message) -> {throw original;}); runMessage(); + assertSame(original,only(Stage.SEND,1).failure); + assertSame(original,only(Stage.DESTINATION,1).failure); + assertTrue(records(Stage.RESPONSE,1).isEmpty()); status(1,Status.ERROR); + } + @Test public void queuedRetryRunsPerAttemptSendAndResponseOnQueueWorker() throws Exception { + create(true, 1); + AtomicInteger calls = new AtomicInteger(); + TestDestinationConnector d=sender(() -> new Response(calls.incrementAndGet()==1?Status.QUEUED:Status.SENT,"reply")); + DestinationConnectorProperties props=((TestConnectorProperties)d.getConnectorProperties()).getDestinationConnectorProperties(); + props.setQueueEnabled(true); props.setSendFirst(false); props.setRegenerateTemplate(true); props.setRetryIntervalMillis(1); runMessage(); + long deadline=System.nanoTime()+TimeUnit.SECONDS.toNanos(5); + while (true) { + if (System.nanoTime()>deadline) fail("queued retry not completed"); + try { status(1,Status.SENT); break; } catch (AssertionError waiting) { Thread.sleep(5); } + } + assertEquals(2,calls.get()); assertEquals(2,records(Stage.SEND,1).size()); assertEquals(2,records(Stage.RESPONSE,1).size()); + for(Recorded record:records(Stage.SEND,1)) assertTrue(record.thread instanceof DestinationConnector.DestinationQueueThread); + assertEquals(Status.QUEUED,records(Stage.SEND,1).get(0).status); + assertEquals(Status.SENT,records(Stage.SEND,1).get(1).status); + List attempts = records(Stage.DESTINATION, 1).stream() + .filter(r -> r.thread instanceof DestinationConnector.DestinationQueueThread).toList(); + assertEquals(2, attempts.size()); + for (int i = 0; i < 2; i++) assertSame(attempts.get(i), records(Stage.SEND, 1).get(i).parent); + } + + @Test public void queuedTransformationBelongsToAcquiredDestinationScope() throws Exception { + create(true, 1); + var destination = sender(() -> new Response(Status.SENT, "reply")); + queueProperties(destination).setIncludeFilterTransformer(true); + runMessage(); awaitQueuedCompletion(); + List coarse = records(Stage.DESTINATION, 1); + assertEquals(2, coarse.size()); + Recorded actual = coarse.stream().filter(r -> r.thread instanceof DestinationConnector.DestinationQueueThread).findFirst().orElseThrow(); + assertSame(actual, only(Stage.TRANSFORM, 1).parent); + assertSame(actual, only(Stage.SEND, 1).parent); + assertSame(actual, only(Stage.RESPONSE, 1).parent); + assertEquals(Status.QUEUED, coarse.get(0).status); + } + + @Test public void interruptedHeldRetryIsObservedBeforeAnySecondSend() throws Exception { + create(true, 1); + AtomicInteger starts = new AtomicInteger(), sends = new AtomicInteger(); + CountDownLatch retry = new CountDownLatch(1); + java.util.concurrent.atomic.AtomicReference retryThread = new java.util.concurrent.atomic.AtomicReference<>(); + var destination = sender(() -> { sends.incrementAndGet(); return new Response(Status.QUEUED, "reply"); }); + queueProperties(destination).setRetryIntervalMillis(30000); + recorder.beforeStage = (stage, message) -> { + if (stage == Stage.DESTINATION && Thread.currentThread() instanceof DestinationConnector.DestinationQueueThread + && starts.incrementAndGet() == 2) { + retryThread.set(Thread.currentThread()); retry.countDown(); + } + }; + runMessage(); assertTrue(retry.await(5, TimeUnit.SECONDS)); retryThread.get().interrupt(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (destination.isQueueThreadRunning()) { + if (System.nanoTime() - deadline >= 0) fail("interrupted queue worker did not exit"); + Thread.sleep(5); + } + List attempts = records(Stage.DESTINATION, 1).stream() + .filter(r -> r.thread instanceof DestinationConnector.DestinationQueueThread).toList(); + assertEquals(2, attempts.size()); assertEquals(1, sends.get()); + assertTrue(attempts.get(1).failure instanceof InterruptedException); + assertEquals(Status.QUEUED, attempts.get(1).status); assertEquals(1, attempts.get(1).closed); + status(1, Status.QUEUED); + } + + private void assertHierarchy(int destinations) { + assertEquals(2 + destinations * 4, recorder.records.size()); + Recorded source = recorder.only(Stage.SOURCE, 0); + assertNull(source.parent); + for (Recorded record : recorder.records) { + if (record != source) { + Recorded expected = record.message.getMetaDataId() == 0 || record.stage == Stage.DESTINATION + ? source : recorder.only(Stage.DESTINATION, record.message.getMetaDataId()); + assertSame("details belong to the actual destination scope", expected, record.parent); + } + assertEquals(1, record.closed); + } + } + + private static final class Recorded { + final Stage stage; + final ConnectorMessage message; + final Recorded parent; + final Thread thread = Thread.currentThread(); + volatile int closed; + Status status; + Throwable failure; + Recorded(Stage stage, ConnectorMessage message, Recorded parent) { this.stage = stage; this.message = message; this.parent = parent; } + } + + private static final class Recorder implements MessageTelemetry.Provider { + final ThreadLocal current = new ThreadLocal<>(); + final List records = new CopyOnWriteArrayList<>(); + final List errors = new CopyOnWriteArrayList<>(); + boolean failCallbacks; + java.util.function.BiConsumer> preparation = (message,map) -> {}; + Runnable sourceStarted = () -> {}; + java.util.function.BiConsumer beforeStage = (stage, message) -> {}; + java.util.function.Consumer afterClose = record -> {}; + public void beforeStore(ConnectorMessage message, Map sourceMap) { preparation.accept(message,sourceMap); } + public Observation start(Stage stage, ConnectorMessage message) { + beforeStage.accept(stage, message); + if (stage == Stage.SOURCE) sourceStarted.run(); + Recorded record = new Recorded(stage, message, current.get()); + records.add(record); + current.set(record); + return new Observation() { + public void status(Status status) { record.status = status; if (failCallbacks) throw new IllegalStateException(); } + public void failed(Throwable failure) { record.failure = failure; if (failCallbacks) throw new IllegalStateException(); } + public void close() { + if (record.thread != Thread.currentThread() || current.get() != record) errors.add("scope restored on wrong thread or in wrong order"); + current.set(record.parent); + if (record.status == null) record.status = message.getStatus(); + try { afterClose.accept(record); } + finally { record.closed++; } + if (failCallbacks) throw new IllegalStateException(); + } + }; + } + public Supplier capture() { + Recorded captured = current.get(); + return () -> { + Recorded previous = current.get(); + current.set(captured); + return () -> { + if (current.get() != captured) errors.add("worker leaked a nested observation"); + current.set(previous); + }; + }; + } + Recorded only(Stage stage, int connector) { + List matches = new ArrayList<>(); + for (Recorded record : records) if (record.stage == stage && record.message.getMetaDataId() == connector) matches.add(record); + assertEquals("one " + stage + " observation for connector " + connector, 1, matches.size()); + return matches.get(0); + } + } +} diff --git a/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestListenerConnectorProperties.java b/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestListenerConnectorProperties.java index b5fcd1950b..83f13fd8f5 100644 --- a/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestListenerConnectorProperties.java +++ b/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestListenerConnectorProperties.java @@ -22,8 +22,8 @@ @SuppressWarnings("serial") public class TestListenerConnectorProperties extends ConnectorProperties implements ListenerConnectorPropertiesInterface, SourceConnectorPropertiesInterface { - private ListenerConnectorProperties listenerConnectorProperties; - private SourceConnectorProperties sourceConnectorProperties; + private ListenerConnectorProperties listenerConnectorProperties = new ListenerConnectorProperties("0"); + private SourceConnectorProperties sourceConnectorProperties = new SourceConnectorProperties(); @Override public SourceConnectorProperties getSourceConnectorProperties() { diff --git a/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestSourceConnector.java b/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestSourceConnector.java index 4ca4e4d660..e9381fcca7 100644 --- a/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestSourceConnector.java +++ b/donkey/src/test/java/com/mirth/connect/donkey/test/util/TestSourceConnector.java @@ -12,6 +12,7 @@ import java.util.ArrayList; import java.util.List; +import com.mirth.connect.donkey.model.channel.ConnectorProperties; import com.mirth.connect.donkey.model.message.RawMessage; import com.mirth.connect.donkey.server.ConnectorTaskException; import com.mirth.connect.donkey.server.channel.ChannelException; @@ -19,7 +20,7 @@ import com.mirth.connect.donkey.server.channel.SourceConnector; public class TestSourceConnector extends SourceConnector { - protected TestConnectorProperties connectorProperties; + protected ConnectorProperties connectorProperties; private List recoveredDispatchResults = new ArrayList(); private boolean isDeployed = false; private List messageIds = new ArrayList(); @@ -38,7 +39,7 @@ public List getMessageIds() { @Override public void onDeploy() { - this.connectorProperties = (TestConnectorProperties) getConnectorProperties(); + this.connectorProperties = getConnectorProperties(); isDeployed = true; } diff --git a/server/src/main/java/com/mirth/connect/server/util/javascript/JavaScriptUtil.java b/server/src/main/java/com/mirth/connect/server/util/javascript/JavaScriptUtil.java index 592b899e5b..a5da6134f1 100644 --- a/server/src/main/java/com/mirth/connect/server/util/javascript/JavaScriptUtil.java +++ b/server/src/main/java/com/mirth/connect/server/util/javascript/JavaScriptUtil.java @@ -71,7 +71,7 @@ public class JavaScriptUtil { private static String serverId = ControllerFactory.getFactory().createConfigurationController().getServerId(); public static T execute(JavaScriptTask task) throws JavaScriptExecutorException, InterruptedException { - Future future = executor.submit(task); + Future future = executor.submit(com.mirth.connect.donkey.server.channel.MessageTelemetry.wrap(task)); try { return future.get(); diff --git a/server/src/test/java/com/mirth/connect/server/util/javascript/JavaScriptUtilTest.java b/server/src/test/java/com/mirth/connect/server/util/javascript/JavaScriptUtilTest.java index 90221a55f8..8a39f9d958 100644 --- a/server/src/test/java/com/mirth/connect/server/util/javascript/JavaScriptUtilTest.java +++ b/server/src/test/java/com/mirth/connect/server/util/javascript/JavaScriptUtilTest.java @@ -13,6 +13,7 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -20,6 +21,14 @@ import java.net.URL; import java.util.HashMap; import java.util.HashSet; +import java.lang.reflect.Field; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import org.junit.After; import org.junit.BeforeClass; @@ -30,6 +39,7 @@ import com.google.inject.Injector; import com.mirth.connect.donkey.model.message.ConnectorMessage; import com.mirth.connect.donkey.model.message.MessageContent; +import com.mirth.connect.donkey.server.channel.MessageTelemetry; import com.mirth.connect.model.codetemplates.ContextType; import com.mirth.connect.server.builders.JavaScriptBuilder; import com.mirth.connect.server.controllers.CodeTemplateController; @@ -154,4 +164,103 @@ public void preprocessorReturningStringYieldsThatString() throws Exception { CompiledScriptCache.getInstance().removeCompiledScript(scriptId); } } + + @Test + public void telemetryContextTransfersThroughActualJavaScriptExecutorAndRestoresAfterFailure() throws Exception { + withTelemetryWorker((context, worker) -> { + worker.submit(() -> context.set("worker prior")).get(5, TimeUnit.SECONDS); + context.set("source context"); + assertEquals("source context", JavaScriptUtil.execute(telemetryTask(() -> { + assertTrue(Thread.currentThread() instanceof MirthJavaScriptThread); + return context.get(); + }))); + RuntimeException original = new RuntimeException("script failure"); + try { + JavaScriptUtil.execute(telemetryTask(() -> { + assertEquals("source context", context.get()); + throw original; + })); + org.junit.Assert.fail("script must fail"); + } catch (JavaScriptExecutorException failure) { assertSame(original, failure.getCause()); } + assertEquals("worker prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + assertEquals("source context", context.get()); + }); + } + + @Test + public void interruptedScriptCallerLeavesRestorationOnScriptWorker() throws Exception { + withTelemetryWorker((context, worker) -> { + worker.submit(() -> context.set("worker prior")).get(5, TimeUnit.SECONDS); + CountDownLatch started = new CountDownLatch(1), finished = new CountDownLatch(1); + AtomicReference unexpected = new AtomicReference<>(); + Thread caller = new Thread(() -> { + context.set("interrupted request"); + try { + JavaScriptUtil.execute(telemetryTask(() -> { + assertEquals("interrupted request", context.get()); + started.countDown(); + new CountDownLatch(1).await(); + return null; + })); + unexpected.set(new AssertionError("expected caller interruption")); + } catch (InterruptedException expected) { + if (!Thread.currentThread().isInterrupted()) unexpected.set(new AssertionError("interrupt flag cleared")); + } catch (Throwable failure) { unexpected.set(failure); } + finally { finished.countDown(); } + }, "telemetry-script-caller"); + try { + caller.start(); + assertTrue(started.await(5, TimeUnit.SECONDS)); + caller.interrupt(); + assertTrue(finished.await(5, TimeUnit.SECONDS)); + assertNull(unexpected.get()); + assertEquals("worker prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + } finally { + caller.interrupt(); + caller.join(5000); + assertFalse(caller.isAlive()); + } + }); + } + + private JavaScriptTask telemetryTask(Callable body) { + return new JavaScriptTask<>(contextFactory(), "Telemetry test") { + @Override public T doCall() throws Exception { return body.call(); } + }; + } + + @FunctionalInterface + private interface TelemetryScenario { void run(ThreadLocal context, ExecutorService worker) throws Exception; } + + private void withTelemetryWorker(TelemetryScenario scenario) throws Exception { + Field field = JavaScriptUtil.class.getDeclaredField("executor"); + field.setAccessible(true); + ExecutorService priorExecutor = (ExecutorService) field.get(null); + ExecutorService worker = Executors.newSingleThreadExecutor(new MirthJavaScriptThreadFactory()); + ThreadLocal context = new ThreadLocal<>(); + AtomicReference scopeError = new AtomicReference<>(); + field.set(null, worker); + try (AutoCloseable registration = MessageTelemetry.install(new MessageTelemetry.Provider() { + public MessageTelemetry.Observation start(MessageTelemetry.Stage stage, ConnectorMessage message) { return null; } + public Supplier capture() { + String captured = context.get(); + return () -> { + String previous = context.get(); + Thread owner = Thread.currentThread(); + context.set(captured); + return () -> { + if (Thread.currentThread() != owner) scopeError.set("scope closed on another thread"); + context.set(previous); + }; + }; + } + })) { + scenario.run(context, worker); + assertNull(scopeError.get()); + } finally { + field.set(null, priorExecutor); + worker.shutdownNow(); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + } + } } diff --git a/server/src/test/java/com/mirth/connect/server/util/javascript/MessageTelemetryScriptTest.java b/server/src/test/java/com/mirth/connect/server/util/javascript/MessageTelemetryScriptTest.java new file mode 100644 index 0000000000..794cb5d47c --- /dev/null +++ b/server/src/test/java/com/mirth/connect/server/util/javascript/MessageTelemetryScriptTest.java @@ -0,0 +1,100 @@ +/* Published under the Mozilla Public License 2.0. */ +package com.mirth.connect.server.util.javascript; + +import static org.junit.Assert.*; +import org.junit.*; +import org.mozilla.javascript.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.*; +import java.util.*; +import java.net.URL; +import java.lang.reflect.Field; +import java.util.function.Supplier; +import com.mirth.connect.donkey.model.message.ConnectorMessage; +import com.mirth.connect.donkey.server.channel.MessageTelemetry; +import com.mirth.connect.donkey.server.channel.MessageTelemetry.*; + +public class MessageTelemetryScriptTest { + private Field field; + private ExecutorService prior, worker; + private AutoCloseable registration; + private final ThreadLocal context = new ThreadLocal<>(); + private final AtomicInteger active = new AtomicInteger(); + private final AtomicReference scopeError = new AtomicReference<>(); + public static class Bridge { + final ThreadLocal context; final CountDownLatch started; + Bridge(ThreadLocal context, CountDownLatch started) { this.context = context; this.started = started; } + public String read() { return context.get(); } + public void started() { started.countDown(); } + } + @BeforeClass public static void setupControllers() { JavaScriptUtilTest.setUpBeforeClass(); } + @Before public void start() throws Exception { + field = JavaScriptUtil.class.getDeclaredField("executor"); field.setAccessible(true); + prior = (ExecutorService) field.get(null); + worker = Executors.newSingleThreadExecutor(new MirthJavaScriptThreadFactory()); + field.set(null, worker); + worker.submit(() -> context.set("worker prior")).get(5, TimeUnit.SECONDS); + registration = MessageTelemetry.install(new Provider() { + public Observation start(Stage s, ConnectorMessage m) { return null; } + public Supplier capture() { + String captured = context.get(); + return () -> { + String previous = context.get(); Thread thread = Thread.currentThread(); + context.set(captured); active.incrementAndGet(); + return () -> { if (thread != Thread.currentThread()) scopeError.set("wrong owner"); context.set(previous); active.decrementAndGet(); }; + }; + } + }); + } + @After public void cleanup() throws Exception { + try { + assertEquals("worker prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + assertEquals(0, active.get()); assertNull(scopeError.get()); + } finally { + registration.close(); field.set(null, prior); worker.shutdownNow(); + assertTrue(worker.awaitTermination(5, TimeUnit.SECONDS)); + } + } + private JavaScriptTask script(String script, CountDownLatch started) { + return new JavaScriptTask(new MirthContextFactory(new URL[0], new HashSet<>(), false), "review script") { + public Object doCall() throws Exception { + Context cx = getContextFactory().enterContext(); + try { + Scriptable scope = cx.initStandardObjects(); + ScriptableObject.putProperty(scope, "bridge", Context.javaToJS(new Bridge(context, started), scope)); + return executeScript(cx.compileString(script, "telemetry-review", 1, null), scope); + } finally { Context.exit(); } + } + }; + } + @Test public void actualRhinoScriptReadsTransferredContextAndRestoresAfterRhinoFailure() throws Exception { + context.set("source context"); + assertEquals("source context", JavaScriptUtil.execute(script("String(bridge.read());", new CountDownLatch(1)))); + try { + JavaScriptUtil.execute(script("if (String(bridge.read()) !== 'source context') throw new Error('wrong context'); throw 'original script failure';", new CountDownLatch(1))); + fail("script must throw"); + } catch (JavaScriptExecutorException failure) { + assertTrue(failure.getCause() instanceof JavaScriptException); + assertEquals("original script failure", ((JavaScriptException) failure.getCause()).getValue()); + } + assertEquals("source context", context.get()); + } + @Test public void realRhinoInfiniteLoopCancellationStopsViaOriginalTaskMonitor() throws Exception { + CountDownLatch started = new CountDownLatch(1), done = new CountDownLatch(1); + AtomicReference problem = new AtomicReference<>(); + Thread caller = new Thread(() -> { + context.set("interrupted context"); + try { + JavaScriptUtil.execute(script("if (String(bridge.read()) !== 'interrupted context') throw new Error('wrong context'); bridge.started(); while (true) {}", started)); + problem.set(new AssertionError("script unexpectedly returned")); + } catch (InterruptedException expected) { if (!Thread.currentThread().isInterrupted()) problem.set(new AssertionError("interrupt cleared")); } + catch (Throwable t) { problem.set(t); } + finally { done.countDown(); } + }); + try { + caller.start(); assertTrue(started.await(5, TimeUnit.SECONDS)); caller.interrupt(); + assertTrue(done.await(5, TimeUnit.SECONDS)); assertNull(problem.get()); + assertEquals("worker prior", worker.submit(context::get).get(5, TimeUnit.SECONDS)); + } finally { caller.interrupt(); caller.join(5000); assertFalse(caller.isAlive()); } + } +}