diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingResourceSharingExtension.kt b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingResourceSharingExtension.kt index 8aae70afb..cc95dca2b 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/AlertingResourceSharingExtension.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/AlertingResourceSharingExtension.kt @@ -9,6 +9,7 @@ import org.opensearch.commons.alerting.model.ScheduledJob.Companion.SCHEDULED_JO import org.opensearch.security.spi.resources.ResourceProvider import org.opensearch.security.spi.resources.ResourceSharingExtension import org.opensearch.security.spi.resources.client.ResourceSharingClient +import java.util.function.Function class AlertingResourceSharingExtension : ResourceSharingExtension { /** @@ -16,10 +17,9 @@ class AlertingResourceSharingExtension : ResourceSharingExtension { * type-specific field paths so the security plugin can distinguish them without a * top-level discriminator on the stored doc: * - * - [ResourceProvider.typeField] — used by the postIndex hook to classify new writes. - * `monitor.type` is present on monitor docs and absent on workflow docs (and vice versa), - * so the framework iterates matching providers and picks the one whose typeField - * resolves non-null. + * - [ResourceProvider.typeField] and [ResourceProvider.matchesResource]: used to classify + * documents. Workflow documents hold `workflow` in `workflow.type`, while the type is + * registered as `alerting-workflow`, so the workflow provider matches on that value. * - [ResourceProvider.ownerNamePath] / [ResourceProvider.ownerBackendRolesPath] — used by * the security plugin's `POST /_plugins/_security/api/resources/migrate` endpoint when * seeding sharing entries for legacy docs, so a single admin call can attribute owners @@ -38,6 +38,8 @@ class AlertingResourceSharingExtension : ResourceSharingExtension { override fun resourceType(): String = ResourceSharingUtils.WORKFLOW_RESOURCE_TYPE override fun resourceIndexName(): String = SCHEDULED_JOBS_INDEX override fun typeField(): String = "workflow.type" + override fun matchesResource(fieldValue: Function): Boolean = + fieldValue.apply(typeField()) == "workflow" override fun ownerNamePath(): String = "/workflow/user/name" override fun ownerBackendRolesPath(): String = "/workflow/user/backend_roles" } diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt index f4165fb38..13f6f09db 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/transport/TransportIndexWorkflowAction.kt @@ -45,7 +45,7 @@ import org.opensearch.alerting.settings.AlertingSettings.Companion.MAX_TRIGGERS_ import org.opensearch.alerting.settings.AlertingSettings.Companion.REQUEST_TIMEOUT import org.opensearch.alerting.settings.DestinationSettings.Companion.ALLOW_LIST import org.opensearch.alerting.util.IndexUtils -import org.opensearch.alerting.util.indexStashed +import org.opensearch.alerting.util.indexWithContext import org.opensearch.alerting.util.isADMonitor import org.opensearch.alerting.util.isQueryLevelMonitor import org.opensearch.alerting.util.use @@ -227,6 +227,8 @@ class TransportIndexWorkflowAction @Inject constructor( } val tenantId = client.threadPool().threadContext.getHeader(AlertingPlugin.TENANT_ID_HEADER) + // Capture before launch: Dispatchers.IO drops the thread context, and the workflow write needs the caller's auth + val storedContext = client.threadPool().threadContext.newStoredContext(false) scope.launch(TenantContext(tenantId)) { try { val triggerCount = transformedRequest.workflow.triggers.size @@ -246,10 +248,6 @@ class TransportIndexWorkflowAction @Inject constructor( client, object : ActionListener { override fun onResponse(response: AcknowledgedResponse) { - // Capture pre-stash context so the write path can restore the auth-user - // transient — the security plugin's ResourceIndexListener needs it to - // record the workflow's resource-share entry with createdBy=. - val storedContext = client.threadPool().threadContext.newStoredContext(false) client.threadPool().threadContext.stashContext().use { IndexWorkflowHandler( client, actionListener, transformedRequest, user, tenantId, storedContext @@ -285,7 +283,7 @@ class TransportIndexWorkflowAction @Inject constructor( private val request: IndexWorkflowRequest, private val user: User?, private val tenantId: String?, - private val storedThreadContext: org.opensearch.common.util.concurrent.ThreadContext.StoredContext? = null, + private val storedThreadContext: org.opensearch.common.util.concurrent.ThreadContext.StoredContext, ) { fun resolveUserAndStart() { scope.launch(TenantContext(tenantId)) { @@ -432,12 +430,9 @@ class TransportIndexWorkflowAction @Inject constructor( .setIfPrimaryTerm(request.primaryTerm) .timeout(indexTimeout) - // Restore the caller's persistent auth header so the shard-level ResourceIndexListener sees - // the caller when recording the workflow's share entry. [Client.indexStashed] stashes - // per-call so the internal-index write isn't gated by the caller's privileges. - storedThreadContext?.restore() try { - val indexResponse: IndexResponse = client.indexStashed(indexRequest) + // The caller's context reaches only this write, where the security plugin records the creator + val indexResponse: IndexResponse = client.indexWithContext(storedThreadContext, indexRequest) val failureReasons = checkShardsFailure(indexResponse) if (failureReasons != null) { log.error("Failed to create workflow: $failureReasons") @@ -600,11 +595,8 @@ class TransportIndexWorkflowAction @Inject constructor( .setIfPrimaryTerm(request.primaryTerm) .timeout(indexTimeout) - // Restore the caller's persistent auth header for the shard-level ResourceIndexListener; - // [Client.indexStashed] stashes per-call so the internal-index write isn't gated by the caller. - storedThreadContext?.restore() try { - val indexResponse: IndexResponse = client.indexStashed(indexRequest) + val indexResponse: IndexResponse = client.indexWithContext(storedThreadContext, indexRequest) val failureReasons = checkShardsFailure(indexResponse) if (failureReasons != null) { actionListener.onFailure( diff --git a/alerting/src/main/kotlin/org/opensearch/alerting/util/SdkClientExtensions.kt b/alerting/src/main/kotlin/org/opensearch/alerting/util/SdkClientExtensions.kt index a7c27852b..1727f11ee 100644 --- a/alerting/src/main/kotlin/org/opensearch/alerting/util/SdkClientExtensions.kt +++ b/alerting/src/main/kotlin/org/opensearch/alerting/util/SdkClientExtensions.kt @@ -65,19 +65,21 @@ suspend fun SdkClient.deleteDataObjectStashed( } /** - * Executes [Client.index] with the caller's [ThreadContext] stashed just before dispatch; the stash - * is closed via the completion callback so it works across the async / coroutine-resume boundary. - * Prefer this over `client.suspendUntil { client.index(req, it) }` when the write must run under - * the plugin (not the caller) but the surrounding coroutine flow must retain the caller's - * persistent auth for the security plugin's [ResourceIndexListener]. + * Executes [Client.index] under [stored], the caller's context, which is set only while the request is dispatched. + * + * The calling thread gets its own context back before the coroutine suspends, so no thread keeps the caller's + * context, whichever thread the coroutine resumes on. The stash before dispatch keeps the caller's persistent + * headers, which the security plugin reads to record the resource's creator. */ -suspend fun Client.indexStashed( - request: IndexRequest, -): IndexResponse { - val stored = this.threadPool().threadContext.stashContext() - return try { - this.suspendUntil { this.index(request, it) } - } finally { - stored.close() +suspend fun Client.indexWithContext(stored: ThreadContext.StoredContext, request: IndexRequest): IndexResponse { + val threadContext = threadPool().threadContext + return suspendUntil { listener -> + val own = threadContext.newStoredContext(false) + stored.restore() + try { + threadContext.stashContext().use { index(request, listener) } + } finally { + own.restore() + } } } diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureResourceSharingMonitorRestApiIT.kt b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureResourceSharingMonitorRestApiIT.kt index e31489c11..83f05772c 100644 --- a/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureResourceSharingMonitorRestApiIT.kt +++ b/alerting/src/test/kotlin/org/opensearch/alerting/resthandler/SecureResourceSharingMonitorRestApiIT.kt @@ -15,14 +15,19 @@ import org.opensearch.alerting.ALERTING_BASE_URI import org.opensearch.alerting.ALERTING_FULL_ACCESS_ROLE import org.opensearch.alerting.AlertingPlugin.Companion.COMMENTS_BASE_URI import org.opensearch.alerting.AlertingRestTestCase +import org.opensearch.alerting.ResourceSharingUtils.MONITOR_RESOURCE_TYPE +import org.opensearch.alerting.ResourceSharingUtils.WORKFLOW_RESOURCE_TYPE import org.opensearch.alerting.makeRequest import org.opensearch.alerting.randomAlert import org.opensearch.alerting.randomQueryLevelMonitor import org.opensearch.alerting.randomQueryLevelTrigger +import org.opensearch.alerting.randomWorkflow import org.opensearch.client.Request import org.opensearch.client.ResponseException import org.opensearch.client.RestClient import org.opensearch.commons.alerting.model.Alert +import org.opensearch.commons.alerting.model.ScheduledJob +import org.opensearch.commons.alerting.model.Workflow import org.opensearch.commons.rest.SecureRestClientBuilder import org.opensearch.core.rest.RestStatus import org.opensearch.test.junit.annotations.TestLogging @@ -38,6 +43,8 @@ import org.opensearch.test.junit.annotations.TestLogging * - revoke removes access * - cross-resource isolation: a share on monitor A does not grant access to monitor B * - third-party isolation: a share to bob does not grant access to carol + * - search returns a monitor only while it is owned or shared + * - monitor and workflow docs keep `all_shared_principals` in step with share, update and revoke * * Runs only when both `security` and `resource_sharing.enabled` system properties are true. */ @@ -257,21 +264,105 @@ class SecureResourceSharingMonitorRestApiIT : AlertingRestTestCase() { // ─── Search DLS ────────────────────────────────────────────────────────────── - fun `test search excludes monitors not owned or shared`() { - val aliceMonitorId = aliceCreatesMonitor().id + fun `test search returns monitors owned or currently shared`() { + val sharedMonitor = aliceCreatesMonitor() + val unsharedMonitorId = aliceCreatesMonitor().id val bobMonitorId = bobCreatesMonitor().id - val body = searchMonitors(bobClient!!) + var body = searchMonitors(bobClient!!) assertTrue("Bob's own monitor missing: $body", body.contains(bobMonitorId)) - assertFalse("Alice's monitor leaked: $body", body.contains(aliceMonitorId)) + assertFalse("Monitor visible to Bob before share: $body", body.contains(sharedMonitor.id)) + + shareResource(aliceClient!!, sharedMonitor.id, READ_ONLY, RS_BOB) + body = searchMonitors(bobClient!!) + assertTrue("Bob's own monitor missing after share: $body", body.contains(bobMonitorId)) + assertTrue("Shared monitor missing: $body", body.contains(sharedMonitor.id)) + assertFalse("Unshared monitor visible to Bob: $body", body.contains(unsharedMonitorId)) + + updateMonitorAs(aliceClient!!, sharedMonitor.copy(name = "renamed")) + awaitStoredPrincipals(sharedMonitor.id, setOf("user:$RS_ALICE", "user:$RS_BOB")) + body = searchMonitors(bobClient!!) + assertTrue("Shared monitor missing after update: $body", body.contains(sharedMonitor.id)) + + revokeResource(aliceClient!!, sharedMonitor.id, RS_BOB) + body = searchMonitors(bobClient!!) + assertTrue("Bob's own monitor missing after revoke: $body", body.contains(bobMonitorId)) + assertFalse("Monitor visible to Bob after revoke: $body", body.contains(sharedMonitor.id)) } - fun `test search includes shared monitor`() { + fun `test search returns workflows owned or currently shared`() { val aliceMonitorId = aliceCreatesMonitor().id - shareResource(aliceClient!!, aliceMonitorId, READ_ONLY, RS_BOB) + val workflowId = createWorkflowWithClient(aliceClient!!, randomWorkflow(monitorIds = listOf(aliceMonitorId))).id + val bobMonitorId = bobCreatesMonitor().id + + // The security plugin writes the workflow's sharing entry and all_shared_principals after create returns. + assertBusy { + val body = searchMonitors(aliceClient!!) + assertTrue("Alice's own monitor missing: $body", body.contains(aliceMonitorId)) + assertTrue("Alice's own workflow missing: $body", body.contains(workflowId)) + } + var body = searchMonitors(bobClient!!) + assertTrue("Bob's own monitor missing: $body", body.contains(bobMonitorId)) + assertFalse("Workflow visible to Bob before share: $body", body.contains(workflowId)) + + shareResource(aliceClient!!, workflowId, READ_ONLY, RS_BOB, WORKFLOW_RESOURCE_TYPE) + body = searchMonitors(bobClient!!) + assertTrue("Shared workflow missing: $body", body.contains(workflowId)) + + revokeResource(aliceClient!!, workflowId, RS_BOB, WORKFLOW_RESOURCE_TYPE) + body = searchMonitors(bobClient!!) + assertTrue("Bob's own monitor missing after revoke: $body", body.contains(bobMonitorId)) + assertFalse("Workflow visible to Bob after revoke: $body", body.contains(workflowId)) + } + + // ─── all_shared_principals on the resource document ───────────────────────── + // DLS and accessible-id listing match the doc's own all_shared_principals, so it must be stored and keyword-indexed. + + fun `test all_shared_principals follows share, update and revoke on monitor`() { + val monitor = aliceCreatesMonitor() + assertPrincipals(monitor.id, setOf("user:$RS_ALICE")) + + shareResource(aliceClient!!, monitor.id, READ_WRITE, RS_BOB) + assertPrincipals(monitor.id, setOf("user:$RS_ALICE", "user:$RS_BOB")) + + // A monitor update replaces the whole document, so the field must be restored after it. + updateMonitorAs(bobClient!!, monitor.copy(name = "renamed-by-bob")) + assertEquals("renamed-by-bob", storedSource(monitor.id).stringMap("monitor")?.get("name")) + assertPrincipals(monitor.id, setOf("user:$RS_ALICE", "user:$RS_BOB")) + + revokeResource(aliceClient!!, monitor.id, RS_BOB) + assertPrincipals(monitor.id, setOf("user:$RS_ALICE"), absent = setOf("user:$RS_BOB")) + } + + fun `test all_shared_principals follows share, update and revoke on workflow`() { + val workflow = aliceCreatesWorkflow() + assertPrincipals(workflow.id, setOf("user:$RS_ALICE")) + + shareResource(aliceClient!!, workflow.id, READ_WRITE, RS_BOB, WORKFLOW_RESOURCE_TYPE) + assertPrincipals(workflow.id, setOf("user:$RS_ALICE", "user:$RS_BOB")) + + // Updated as the owner: a workflow update by bob would also need access to its delegate monitor. + updateWorkflowAs(aliceClient!!, workflow.copy(name = "renamed")) + assertEquals("renamed", storedSource(workflow.id).stringMap("workflow")?.get("name")) + assertPrincipals(workflow.id, setOf("user:$RS_ALICE", "user:$RS_BOB")) + + revokeResource(aliceClient!!, workflow.id, RS_BOB, WORKFLOW_RESOURCE_TYPE) + assertPrincipals(workflow.id, setOf("user:$RS_ALICE"), absent = setOf("user:$RS_BOB")) + } + + fun `test workflow update by a read-write shared user keeps the owner`() { + val monitorId = aliceCreatesMonitor().id + val workflow = createWorkflowWithClient(aliceClient!!, randomWorkflow(monitorIds = listOf(monitorId))) + waitForResourceSharingEntry(workflow.id) + // Bob also needs its delegate monitor to update the workflow + shareResource(aliceClient!!, monitorId, READ_WRITE, RS_BOB) + shareResource(aliceClient!!, workflow.id, READ_WRITE, RS_BOB, WORKFLOW_RESOURCE_TYPE) + + updateWorkflowAs(bobClient!!, workflow.copy(name = "renamed-by-bob")) - val body = searchMonitors(bobClient!!) - assertTrue("Shared monitor missing from search: $body", body.contains(aliceMonitorId)) + assertEquals("renamed-by-bob", storedSource(workflow.id).stringMap("workflow")?.get("name")) + assertEquals(RS_ALICE, sharingRecordOwner(workflow.id)) + assertPrincipals(workflow.id, setOf("user:$RS_ALICE", "user:$RS_BOB")) } // ─── Subordinate resource: alerts ──────────────────────────────────────────── @@ -418,6 +509,80 @@ class SecureResourceSharingMonitorRestApiIT : AlertingRestTestCase() { private fun bobCreatesMonitor() = createMonitorAs(bobClient!!, sampleMonitor()) + /** Creates a workflow as alice over one of her monitors and waits for its sharing entry. */ + private fun aliceCreatesWorkflow(): Workflow { + val monitorId = aliceCreatesMonitor().id + val workflow = createWorkflowWithClient(aliceClient!!, randomWorkflow(monitorIds = listOf(monitorId))) + waitForResourceSharingEntry(workflow.id) + return workflow + } + + /** PUTs a workflow as [client] without the base helper's follow-up GET, which runs as admin. */ + private fun updateWorkflowAs(client: RestClient, workflow: Workflow) { + val response = client.makeRequest("PUT", "${workflow.relativeUrl()}?refresh=true", emptyMap(), workflow.toHttpEntity()) + assertEquals("Unable to update a workflow", RestStatus.OK.status, response.statusLine.statusCode) + } + + /** Asserts [docId] stores exactly [expected] in `all_shared_principals`, and term queries match each and none of [absent]. */ + private fun assertPrincipals(docId: String, expected: Set, absent: Set = emptySet()) { + awaitStoredPrincipals(docId, expected) + expected.forEach { principal -> + assertEquals("Term query on all_shared_principals=$principal for $docId", 1, countMatching(docId, principal)) + } + // user:nobody proves the term query can miss, so the matches above come from the indexed field. + (absent + "user:nobody").forEach { principal -> + assertEquals("Term query on absent principal $principal for $docId", 0, countMatching(docId, principal)) + } + } + + /** Waits until [docId] stores exactly [expected] in `all_shared_principals`, then refreshes the index for search. */ + private fun awaitStoredPrincipals(docId: String, expected: Set) { + // Create and update write the field asynchronously, from the security plugin's index listener. + assertBusy { + assertEquals("all_shared_principals in _source of $docId", expected.sorted(), storedPrincipals(docId).sorted()) + } + adminClient().makeRequest("POST", "/${ScheduledJob.SCHEDULED_JOBS_INDEX}/_refresh") + } + + private fun storedSource(docId: String): Map { + val response = adminClient().makeRequest("GET", "/${ScheduledJob.SCHEDULED_JOBS_INDEX}/_doc/$docId") + return response.asMap()["_source"] as Map + } + + private fun sharingRecordOwner(docId: String): String? { + val source = adminClient().makeRequest("GET", "/$configSharingIndex/_doc/$docId").asMap()["_source"] as Map<*, *> + return (source["created_by"] as? Map<*, *>)?.get("user") as? String + } + + /** Returns the stored values as a list, so a duplicate value fails the comparison. */ + private fun storedPrincipals(docId: String): List { + val raw = storedSource(docId)["all_shared_principals"] ?: return emptyList() + val values = raw as? List<*> ?: throw AssertionError("all_shared_principals is not a list on $docId: $raw") + return values.map { it as String } + } + + private fun countMatching(docId: String, principal: String): Int { + val query = """ + { + "query": { + "bool": { + "filter": [ + { "ids": { "values": ["$docId"] } }, + { "term": { "all_shared_principals": "$principal" } } + ] + } + } + } + """.trimIndent() + val response = adminClient().makeRequest( + "POST", + "/${ScheduledJob.SCHEDULED_JOBS_INDEX}/_count", + emptyMap(), + StringEntity(query, ContentType.APPLICATION_JSON) + ) + return (response.asMap()["count"] as Number).toInt() + } + /** * POSTs a monitor as [client] and parses the id out of the response directly. * The base test's [createMonitorWithClient] does a follow-up GET with the default admin client, @@ -564,13 +729,19 @@ class SecureResourceSharingMonitorRestApiIT : AlertingRestTestCase() { } } - private fun shareResource(client: RestClient, resourceId: String, accessLevel: String, user: String) { + private fun shareResource( + client: RestClient, + resourceId: String, + accessLevel: String, + user: String, + resourceType: String = MONITOR_RESOURCE_TYPE + ) { val request = Request("PUT", "/_plugins/_security/api/resource/share") request.setJsonEntity( """ { "resource_id": "$resourceId", - "resource_type": "monitor", + "resource_type": "$resourceType", "share_with": { "$accessLevel": { "users": ["$user"] } } } """.trimIndent() @@ -581,10 +752,7 @@ class SecureResourceSharingMonitorRestApiIT : AlertingRestTestCase() { // security plugin, but any secondary queries that hit the sharing index via a search // (for example `getAccessibleResourceIds`) may still miss until the shard's search view // catches up. Force a refresh so `getAccessibleResourceIds` picks up the new entry. - try { - adminClient().performRequest(Request("POST", "/$configSharingIndex/_refresh")) - } catch (_: Exception) { - } + adminClient().performRequest(Request("POST", "/$configSharingIndex/_refresh")) } /** @@ -593,13 +761,18 @@ class SecureResourceSharingMonitorRestApiIT : AlertingRestTestCase() { * enumerate all access levels the user might be shared at — we pass all three since tests * don't always know which level was granted. */ - private fun revokeResource(client: RestClient, resourceId: String, user: String) { + private fun revokeResource( + client: RestClient, + resourceId: String, + user: String, + resourceType: String = MONITOR_RESOURCE_TYPE + ) { val request = Request("PATCH", "/_plugins/_security/api/resource/share") request.setJsonEntity( """ { "resource_id": "$resourceId", - "resource_type": "monitor", + "resource_type": "$resourceType", "revoke": { "$READ_ONLY": { "users": ["$user"] }, "$READ_WRITE": { "users": ["$user"] }, diff --git a/alerting/src/test/kotlin/org/opensearch/alerting/util/SdkClientExtensionsTests.kt b/alerting/src/test/kotlin/org/opensearch/alerting/util/SdkClientExtensionsTests.kt new file mode 100644 index 000000000..ced2cfa7c --- /dev/null +++ b/alerting/src/test/kotlin/org/opensearch/alerting/util/SdkClientExtensionsTests.kt @@ -0,0 +1,75 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.alerting.util + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import org.mockito.ArgumentMatchers.any +import org.mockito.Mockito.doAnswer +import org.mockito.Mockito.mock +import org.mockito.Mockito.`when` +import org.opensearch.action.index.IndexRequest +import org.opensearch.action.index.IndexResponse +import org.opensearch.common.settings.Settings +import org.opensearch.common.util.concurrent.ThreadContext +import org.opensearch.core.action.ActionListener +import org.opensearch.test.OpenSearchTestCase +import org.opensearch.threadpool.ThreadPool +import org.opensearch.transport.client.Client +import java.util.concurrent.atomic.AtomicReference + +class SdkClientExtensionsTests : OpenSearchTestCase() { + + private val callerHeader = "caller" + + /** + * The coroutine starts on this thread and resumes on the thread that answers the write, as it can on + * Dispatchers.IO. Neither thread may keep the caller's context afterwards. + */ + fun `test indexWithContext applies the caller's context to the write and leaves it on no thread`() { + val threadContext = ThreadContext(Settings.EMPTY) + val threadPool = mock(ThreadPool::class.java) + `when`(threadPool.threadContext).thenReturn(threadContext) + val client = mock(Client::class.java) + `when`(client.threadPool()).thenReturn(threadPool) + + val caller = threadContext.stashContext().use { + threadContext.putPersistent(callerHeader, "alice") + threadContext.newStoredContext(false) + } + + val seenByWrite = AtomicReference() + val pending = AtomicReference>() + doAnswer { + seenByWrite.set(threadContext.getPersistent(callerHeader)) + pending.set(it.getArgument(1)) + null + }.`when`(client).index(any(IndexRequest::class.java), any()) + + val response = mock(IndexResponse::class.java) + val result = AtomicReference() + val seenByStarter = AtomicReference("unset") + val seenByResponder = AtomicReference("unset") + runBlocking { + // Unconfined runs the coroutine here until it suspends, and resumes it on the thread that answers + launch(Dispatchers.Unconfined) { result.set(client.indexWithContext(caller, IndexRequest("index"))) } + seenByStarter.set(threadContext.getPersistent(callerHeader)) + + val responder = Thread { + pending.get().onResponse(response) + seenByResponder.set(threadContext.getPersistent(callerHeader)) + } + responder.start() + responder.join() + } + + assertEquals("alice", seenByWrite.get()) + assertSame(response, result.get()) + assertNull("the starting thread kept the caller's context", seenByStarter.get()) + assertNull("the answering thread took the caller's context", seenByResponder.get()) + } +}