diff --git a/app/src/androidTest/java/dev/threadline/ConnectionFormRetentionTest.kt b/app/src/androidTest/java/dev/threadline/ConnectionFormRetentionTest.kt index 4d3b986..ab5271b 100644 --- a/app/src/androidTest/java/dev/threadline/ConnectionFormRetentionTest.kt +++ b/app/src/androidTest/java/dev/threadline/ConnectionFormRetentionTest.kt @@ -584,6 +584,10 @@ class ConnectionFormRetentionTest { .performClick() compose.waitForIdle() compose.onNodeWithText("printf saved").assertExists() + compose.onNodeWithText( + "While connected, only completed commands are saved. " + + "A crash can lose the unfinished command.", + ).assertExists() compose.onNodeWithTag(TranscriptHistoryTags.OUTPUT_PREFIX + "command-session-1") .assertExists() compose.onNodeWithText("Succeeded · exit 0").assertExists() @@ -609,6 +613,25 @@ class ConnectionFormRetentionTest { compose.onAllNodesWithTag(TranscriptHistoryTags.OPEN).assertCountEquals(0) } + @Test + fun transcriptSaveFailureDoesNotClaimTheSessionEnded() { + compose.setContent { + MaterialTheme { + TranscriptHistorySection( + sessions = emptyList(), + saveFailed = true, + onLoad = { error("No saved session") }, + onDelete = {}, + onClearAll = {}, + ) + } + } + compose.onNodeWithText( + "The latest transcript could not be saved. " + + "Recent commands may be missing from history.", + ).assertExists() + } + @Test fun savedKeySelectionLoadsCredentialAndWipesTemporaryPassphrase() { val savedKey = ImportedPrivateKeyMetadata( @@ -1338,7 +1361,7 @@ private fun transcriptSummary( port = 2222, username = "threadline", startedAtMillis = 1, - endedAtMillis = 2, + savedAtMillis = 2, turnsTruncated = false, turnCount = 1, ) diff --git a/app/src/androidTest/java/dev/threadline/core/session/AndroidTranscriptCheckpointProcessDeathTest.kt b/app/src/androidTest/java/dev/threadline/core/session/AndroidTranscriptCheckpointProcessDeathTest.kt new file mode 100644 index 0000000..0791ac7 --- /dev/null +++ b/app/src/androidTest/java/dev/threadline/core/session/AndroidTranscriptCheckpointProcessDeathTest.kt @@ -0,0 +1,138 @@ +package dev.threadline.core.session + +import android.os.Bundle +import android.os.Process +import androidx.test.ext.junit.runners.AndroidJUnit4 +import androidx.test.platform.app.InstrumentationRegistry +import dev.threadline.SessionRuntime +import dev.threadline.core.model.ConnectionRequest +import dev.threadline.core.model.HostEndpoint +import dev.threadline.core.model.HostKeyDecision +import dev.threadline.core.model.HostProfile +import dev.threadline.core.model.SessionCredential +import dev.threadline.core.model.SessionState +import dev.threadline.core.shell.CommandSubmissionResult +import dev.threadline.core.shell.StructuredShellState +import dev.threadline.core.transcript.CommandStatus +import java.io.File +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.flow.filterIsInstance +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Assume.assumeTrue +import org.junit.Test +import org.junit.runner.RunWith + +@RunWith(AndroidJUnit4::class) +class AndroidTranscriptCheckpointProcessDeathTest { + // The fixture runner kills the preparing process, then verifies in a new one. + @Test + fun completedHistorySurvivesProcessDeathWithoutResumingConnection() = runBlocking { + val arguments = InstrumentationRegistry.getArguments() + val phase = arguments.getString("threadlineCheckpointPhase") + assumeTrue("Run through test-android-transcript-process-death.sh", phase != null) + val ephemeral = arguments.getString("threadlineEphemeral") == "true" + val manager = SessionRuntime.manager + val history = SessionRuntime.transcriptHistory + + assertEquals(SessionState.Disconnected, manager.state.value) + assertEquals(StructuredShellState.Inactive, manager.structuredState.value) + assertTrue(manager.transcriptState.value.turns.isEmpty()) + + if (phase == "verify") { + val sessions = history.sessions.first() + if (ephemeral) { + assertTrue(sessions.isEmpty()) + } else { + val saved = history.load(sessions.single().id) + val turn = saved.turns.single().turn + assertEquals("printf 'checkpoint-completed\\n'", turn.command) + assertEquals("checkpoint-completed\n", turn.output.plainText) + assertEquals(CommandStatus.SUCCEEDED, turn.status) + assertTrue(turn.completedAtMillis != null) + } + history.clearAll() + return@runBlocking + } + + assertEquals("prepare", phase) + history.clearAll() + val context = InstrumentationRegistry.getInstrumentation().targetContext + val keyFile = File(context.filesDir, "threadline-fixture-private-key") + val keyBytes = keyFile.readBytes() + keyFile.delete() + val credential = try { + SessionCredential.PrivateKey.from(keyBytes, passphrase = null) + } finally { + keyBytes.fill(0) + } + try { + assertTrue( + manager.prepareConnection( + ConnectionRequest( + profile = HostProfile( + displayName = "Checkpoint fixture", + endpoint = HostEndpoint("10.0.2.2", 2222), + username = "threadline", + ), + credential = credential, + ephemeral = ephemeral, + ), + ), + ) + assertTrue(manager.connectPrepared()) + val connection = withTimeout(20_000) { + manager.state.first { + it is SessionState.AwaitingHostKey || it is SessionState.Connected || + it is SessionState.Failed + } + } + if (connection is SessionState.AwaitingHostKey) { + assertEquals( + arguments.getString("threadlineFixtureHostKeyFingerprint"), + connection.prompt.fingerprint, + ) + assertTrue(manager.resolveHostKey(HostKeyDecision.ACCEPT_AND_SAVE)) + } + withTimeout(20_000) { + manager.state.filterIsInstance().first() + manager.structuredState.filterIsInstance().first() + } + } finally { + credential.clear() + } + + val completed = manager.submitCommand("printf 'checkpoint-completed\\n'") + as CommandSubmissionResult.Accepted + withTimeout(20_000) { + manager.structuredState.filterIsInstance() + .first { it.lastCommand?.id == completed.commandId } + if (!ephemeral) history.sessions.first { it.singleOrNull()?.turnCount == 1 } + } + val unfinished = manager.submitCommand("printf 'checkpoint-unfinished\\n'; sleep 120") + as CommandSubmissionResult.Accepted + withTimeout(20_000) { + manager.transcriptState.first { transcript -> + transcript.turns.firstOrNull { it.id == unfinished.commandId } + ?.output?.plainText == "checkpoint-unfinished\n" + } + } + assertTrue(manager.state.value is SessionState.Connected) + assertTrue(!manager.transcriptSaveFailed.value) + val saved = history.sessions.first() + assertEquals(if (ephemeral) 0 else 1, saved.size) + if (!ephemeral) assertEquals(1, history.load(saved.single().id).turns.size) + + InstrumentationRegistry.getInstrumentation().sendStatus( + 0, + Bundle().apply { + putString("stream", "THREADLINE_CHECKPOINT_READY pid=${Process.myPid()}\n") + }, + ) + // No disconnect or database close: the host must kill this live process. + withTimeout(60_000) { awaitCancellation() } + } +} diff --git a/app/src/androidTest/java/dev/threadline/data/transcript/RoomTranscriptHistoryStoreTest.kt b/app/src/androidTest/java/dev/threadline/data/transcript/RoomTranscriptHistoryStoreTest.kt index 7264fe4..a4931de 100644 --- a/app/src/androidTest/java/dev/threadline/data/transcript/RoomTranscriptHistoryStoreTest.kt +++ b/app/src/androidTest/java/dev/threadline/data/transcript/RoomTranscriptHistoryStoreTest.kt @@ -45,7 +45,7 @@ class RoomTranscriptHistoryStoreTest { } @Test - fun archiveSurvivesReopenWithOrderedChunksAndBoundedNewestData() = runBlocking { + fun checkpointsAndFinalArchiveSurviveReopenWithBoundedNewestData() = runBlocking { val store = store() val retainedOutput = "x".repeat(RoomTranscriptHistoryStore.OUTPUT_CHUNK_CHARACTERS - 1) + @@ -80,6 +80,9 @@ class RoomTranscriptHistoryStoreTest { }, ) } + turns.indices.forEach { index -> + store.save(archive("persistent-session", index.toLong(), turns.take(index + 1))) + } store.save(archive("persistent-session", 100, turns)) database.close() @@ -87,6 +90,7 @@ class RoomTranscriptHistoryStoreTest { val reopenedStore = store() val summary = reopenedStore.sessions.first().single() assertEquals("persistent-session", summary.id) + assertEquals(100L, summary.savedAtMillis) assertEquals(RoomTranscriptHistoryStore.MAXIMUM_TURNS_PER_SESSION, summary.turnCount) assertTrue(summary.turnsTruncated) @@ -129,6 +133,12 @@ class RoomTranscriptHistoryStoreTest { assertNull(database.transcriptArchives().findSession("session-0")) assertEquals("session-20", retained.first().id) + store.save(archive("session-20", 21, listOf(turn(20), turn(21)))) + assertEquals(RoomTranscriptHistoryStore.MAXIMUM_SESSIONS, store.sessions.first().size) + assertEquals(2, store.load("session-20").turns.size) + assertEquals(2, database.transcriptArchives().findTurns("session-20").size) + assertEquals(2, database.transcriptArchives().findChunks("session-20").size) + store.delete("session-10") assertNull(database.transcriptArchives().findSession("session-10")) assertTrue(database.transcriptArchives().findSession("session-11") != null) @@ -164,7 +174,7 @@ class RoomTranscriptHistoryStoreTest { private fun archive( id: String, - endedAtMillis: Long, + savedAtMillis: Long, turns: List, ) = TranscriptSessionArchive( id = id, @@ -174,7 +184,7 @@ class RoomTranscriptHistoryStoreTest { username = "threadline", ), startedAtMillis = 1, - endedAtMillis = endedAtMillis, + savedAtMillis = savedAtMillis, transcript = CommandTranscriptState(turns = turns), ) diff --git a/app/src/main/java/dev/threadline/TranscriptHistory.kt b/app/src/main/java/dev/threadline/TranscriptHistory.kt index 2f55286..bd2ca8a 100644 --- a/app/src/main/java/dev/threadline/TranscriptHistory.kt +++ b/app/src/main/java/dev/threadline/TranscriptHistory.kt @@ -68,7 +68,8 @@ internal fun TranscriptHistorySection( if (saveFailed || error != null) { Card(modifier = Modifier.fillMaxWidth()) { Text( - text = error ?: "The last session ended, but its transcript could not be saved.", + text = error ?: "The latest transcript could not be saved. " + + "Recent commands may be missing from history.", color = MaterialTheme.colorScheme.error, modifier = Modifier.padding(16.dp), ) @@ -120,7 +121,7 @@ internal fun TranscriptHistorySection( fontFamily = FontFamily.Monospace, ) Text( - "${formatTranscriptTimestamp(session.endedAtMillis)} · " + + "${formatTranscriptTimestamp(session.savedAtMillis)} · " + "${session.turnCount} " + if (session.turnCount == 1) "turn" else "turns", style = MaterialTheme.typography.bodySmall, @@ -222,12 +223,17 @@ internal fun TranscriptHistorySection( style = MaterialTheme.typography.bodySmall, ) Text( - "Ended " + - formatTranscriptTimestamp(session.summary.endedAtMillis) + + "Saved " + + formatTranscriptTimestamp(session.summary.savedAtMillis) + ". Saved output is plain text; terminal styling and " + "links are inert.", style = MaterialTheme.typography.bodySmall, ) + Text( + "While connected, only completed commands are saved. " + + "A crash can lose the unfinished command.", + style = MaterialTheme.typography.bodySmall, + ) } } if (session.summary.turnsTruncated) item { diff --git a/app/src/main/java/dev/threadline/core/session/SessionManager.kt b/app/src/main/java/dev/threadline/core/session/SessionManager.kt index d3d19b7..00ff46b 100644 --- a/app/src/main/java/dev/threadline/core/session/SessionManager.kt +++ b/app/src/main/java/dev/threadline/core/session/SessionManager.kt @@ -58,6 +58,8 @@ import kotlinx.coroutines.flow.debounce import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock enum class ReplySubmissionResult { SENT, @@ -89,6 +91,7 @@ class SessionManager( private val decisionLock = Any() private val structuredLock = Any() private val transcriptArchiveLock = Any() + private val transcriptSaveMutex = Mutex() private val resizeRequests = MutableSharedFlow( extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST, @@ -667,6 +670,7 @@ class SessionManager( } is ProtocolStreamItem.Lifecycle -> { + var commandCompleted = false val pendingBytes = synchronized(structuredLock) { val pending = if (item.event is ShellLifecycleEvent.CommandStarted) { internalInputEchoFilter.cancel() @@ -674,6 +678,8 @@ class SessionManager( byteArrayOf() } if (structuredContext === context) { + commandCompleted = item.event is ShellLifecycleEvent.CommandEnded && + commandTranscript.state.value.activeCommandId == item.event.commandId commandTranscript.lifecycle(item.event) val next = structuredStateMachine.apply( StructuredShellEvent.Lifecycle(item.event), @@ -692,6 +698,7 @@ class SessionManager( pending } if (pendingBytes.isNotEmpty()) terminal.receive(pendingBytes) + if (commandCompleted) checkpointTranscriptSession(context) } } } @@ -738,24 +745,53 @@ class SessionManager( synchronized(transcriptArchiveLock) { val active = activeTranscriptSession ?: return@synchronized null activeTranscriptSession = null - val startedAtMillis = active.startedAtMillis - if ( - active.ephemeral || - startedAtMillis == null || - commandTranscript.state.value.turns.isEmpty() - ) { - return@synchronized null + transcriptArchive(active, completedOnly = false) + } + + private suspend fun checkpointTranscriptSession(context: StructuredShellContext) { + transcriptSaveMutex.withLock { + // Capture after acquiring the save lock so a closed or replaced session + // cannot enqueue an older snapshot after its final archive. + val archive = synchronized(structuredLock) { + if (structuredContext !== context) return@synchronized null + synchronized(transcriptArchiveLock) { + activeTranscriptSession?.let { transcriptArchive(it, completedOnly = true) } + } } - TranscriptSessionArchive( - id = active.id, - profile = active.profile, - startedAtMillis = startedAtMillis, - endedAtMillis = clockMillis(), - transcript = commandTranscript.state.value, - ) + saveTranscriptArchive(archive) } + } + + private fun transcriptArchive( + active: ActiveTranscriptSession, + completedOnly: Boolean, + ): TranscriptSessionArchive? { + val startedAtMillis = active.startedAtMillis ?: return null + if (active.ephemeral) return null + val transcript = commandTranscript.state.value.let { state -> + if (completedOnly) { + state.copy( + turns = state.turns.filter { it.completedAtMillis != null }, + activeCommandId = null, + ) + } else { + state + } + } + if (transcript.turns.isEmpty()) return null + return TranscriptSessionArchive( + id = active.id, + profile = active.profile, + startedAtMillis = startedAtMillis, + savedAtMillis = clockMillis(), + transcript = transcript, + ) + } + + private suspend fun persistTranscriptArchive(archive: TranscriptSessionArchive?) = + transcriptSaveMutex.withLock { saveTranscriptArchive(archive) } - private suspend fun persistTranscriptArchive(archive: TranscriptSessionArchive?) { + private suspend fun saveTranscriptArchive(archive: TranscriptSessionArchive?) { if (archive == null) return try { transcriptArchiveSink.save(archive) diff --git a/app/src/main/java/dev/threadline/core/transcript/TranscriptArchive.kt b/app/src/main/java/dev/threadline/core/transcript/TranscriptArchive.kt index 0f3fd7a..93199f6 100644 --- a/app/src/main/java/dev/threadline/core/transcript/TranscriptArchive.kt +++ b/app/src/main/java/dev/threadline/core/transcript/TranscriptArchive.kt @@ -6,7 +6,7 @@ data class TranscriptSessionArchive( val id: String, val profile: HostProfile, val startedAtMillis: Long, - val endedAtMillis: Long, + val savedAtMillis: Long, val transcript: CommandTranscriptState, ) diff --git a/app/src/main/java/dev/threadline/data/transcript/RoomTranscriptHistoryStore.kt b/app/src/main/java/dev/threadline/data/transcript/RoomTranscriptHistoryStore.kt index e41ae5e..89e3a56 100644 --- a/app/src/main/java/dev/threadline/data/transcript/RoomTranscriptHistoryStore.kt +++ b/app/src/main/java/dev/threadline/data/transcript/RoomTranscriptHistoryStore.kt @@ -27,7 +27,7 @@ internal data class SavedTranscriptSessionSummary( val port: Int, val username: String, val startedAtMillis: Long, - val endedAtMillis: Long, + val savedAtMillis: Long, val turnsTruncated: Boolean, val turnCount: Int, ) @@ -62,7 +62,8 @@ internal class RoomTranscriptHistoryStore( port = archive.profile.endpoint.port, username = archive.profile.username, startedAtMillis = archive.startedAtMillis, - endedAtMillis = archive.endedAtMillis, + // Keep the legacy column for both checkpoints and finalized archives. + endedAtMillis = archive.savedAtMillis, turnsTruncated = retainedTurns.size < archive.transcript.turns.size, ) val turns = retainedTurns.mapIndexed { index, turn -> @@ -209,7 +210,7 @@ private fun TranscriptSessionSummaryRow.toSummary() = SavedTranscriptSessionSumm port = port, username = username, startedAtMillis = startedAtMillis, - endedAtMillis = endedAtMillis, + savedAtMillis = endedAtMillis, turnsTruncated = turnsTruncated, turnCount = turnCount, ) diff --git a/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt b/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt index 10896c6..bac5d99 100644 --- a/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt +++ b/app/src/test/java/dev/threadline/core/session/SessionManagerIoTest.kt @@ -29,6 +29,7 @@ import dev.threadline.core.transcript.CommandStatus import dev.threadline.core.transcript.TranscriptArchiveSink import dev.threadline.core.transcript.TranscriptSessionArchive import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow @@ -37,6 +38,7 @@ import kotlinx.coroutines.flow.filterIsInstance import kotlinx.coroutines.flow.first import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.withContext import org.junit.Assert.assertArrayEquals import org.junit.Assert.assertEquals import org.junit.Assert.assertTrue @@ -259,8 +261,8 @@ class SessionManagerIoTest { assertEquals("prompt: ", manager.transcriptState.value.turns.single().output.plainText) manager.disconnect() - withTimeout(2_000) { while (archiveSink.archives.isEmpty()) delay(10) } - val archived = archiveSink.archives.single().transcript.turns.single() + withTimeout(2_000) { manager.state.first { it is SessionState.Disconnected } } + val archived = archiveSink.archives.last().transcript.turns.single() assertEquals("read -r reply", archived.command) assertEquals("prompt: ", archived.output.plainText) } @@ -545,12 +547,16 @@ class SessionManagerIoTest { } @Test - fun `durable session archives one completed snapshot on disconnect`() = runBlocking { + fun `durable session checkpoints completed turns before disconnect`() = runBlocking { val session = RecordingSession() val archiveSink = RecordingTranscriptArchiveSink() val nonce = SessionNonce("0123456789abcdef0123456789abcdef") val commandIds = ArrayDeque( - listOf(CommandId("bootstrap-probe"), CommandId("saved-command")), + listOf( + CommandId("bootstrap-probe"), + CommandId("saved-command"), + CommandId("unfinished-command"), + ), ) var now = 100L val manager = SessionManager( @@ -566,13 +572,14 @@ class SessionManagerIoTest { manager.prepareConnection(fixtureRequest()) manager.connectPrepared() - withTimeout(2_000) { + withTimeout(5_000) { manager.state.filterIsInstance().first() } + session.awaitSent(1) session.output.send( lifecycleBytes(nonce, CommandId("bootstrap-probe"), 0, "/home/threadline"), ) - withTimeout(2_000) { + withTimeout(5_000) { manager.structuredState.filterIsInstance().first() } now = 125L @@ -580,6 +587,7 @@ class SessionManagerIoTest { CommandSubmissionResult.Accepted(CommandId("saved-command")), manager.submitCommand("printf saved"), ) + session.awaitSent(2) session.output.send( lifecycleBytes( nonce = nonce, @@ -589,28 +597,116 @@ class SessionManagerIoTest { output = "saved output\r\n", ), ) - withTimeout(2_000) { + withTimeout(5_000) { manager.transcriptState.first { it.turns.singleOrNull()?.status == CommandStatus.SUCCEEDED } } + withTimeout(5_000) { while (archiveSink.archives.isEmpty()) delay(10) } + val checkpoint = archiveSink.archives.single() + assertTrue(manager.state.value is SessionState.Connected) + assertEquals(125L, checkpoint.savedAtMillis) + assertEquals("saved output\n", checkpoint.transcript.turns.single().output.plainText) + + manager.submitCommand("sleep 30") + session.awaitSent(3) + session.output.send( + ("\u001b]777;threadline;${nonce.value};start;unfinished-command\u0007" + + "\u001b]777;threadline;${nonce.value};output;unfinished-command\u0007" + + "unfinished output").encodeToByteArray(), + ) + withTimeout(5_000) { + manager.transcriptState.first { + it.turns.last().output.plainText == "unfinished output" + } + } + assertEquals(1, archiveSink.archives.size) + assertEquals(listOf("printf saved"), checkpoint.transcript.turns.map { it.command }) + assertEquals(null, checkpoint.transcript.activeCommandId) + now = 200L manager.disconnect() - withTimeout(2_000) { + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } - val archived = archiveSink.archives.single() + assertEquals(2, archiveSink.archives.size) + val archived = archiveSink.archives.last() assertEquals("saved-session", archived.id) assertEquals(100L, archived.startedAtMillis) - assertEquals(200L, archived.endedAtMillis) + assertEquals(200L, archived.savedAtMillis) assertEquals("Fixture", archived.profile.displayName) - assertEquals("printf saved", archived.transcript.turns.single().command) - assertEquals("saved output\n", archived.transcript.turns.single().output.plainText) + assertEquals("printf saved", archived.transcript.turns.first().command) + assertEquals("saved output\n", archived.transcript.turns.first().output.plainText) + assertEquals(2, archived.transcript.turns.size) + assertEquals(CommandStatus.DISCONNECTED, archived.transcript.turns.last().status) + assertEquals("unfinished output", archived.transcript.turns.last().output.plainText) assertTrue(!manager.transcriptSaveFailed.value) } + @Test + fun `slow checkpoint cannot overwrite connection loss archive`() = runBlocking { + val session = RecordingSession() + val checkpointStarted = CompletableDeferred() + val releaseCheckpoint = CompletableDeferred() + val archives = CopyOnWriteArrayList() + val nonce = SessionNonce("0123456789abcdef0123456789abcdef") + val commandIds = ArrayDeque( + listOf( + CommandId("bootstrap-probe"), + CommandId("saved-command"), + CommandId("unfinished-command"), + ), + ) + val manager = SessionManager( + adapter = ImmediateAdapter(session), + knownHostStore = EmptyKnownHostStore, + terminal = FakeTerminal, + sessionNonceFactory = { nonce }, + commandIdFactory = { commandIds.removeFirst() }, + transcriptArchiveSink = TranscriptArchiveSink { archive -> + if (archive.transcript.turns.size == 1) { + checkpointStarted.complete(Unit) + withContext(NonCancellable) { releaseCheckpoint.await() } + } + archives += archive + }, + ) + try { + manager.prepareConnection(fixtureRequest()) + manager.connectPrepared() + session.awaitSent(1) + session.output.send(lifecycleBytes(nonce, CommandId("bootstrap-probe"), 0, "/tmp")) + withTimeout(5_000) { + manager.structuredState.filterIsInstance().first() + } + manager.submitCommand("printf saved") + session.awaitSent(2) + session.output.send( + lifecycleBytes(nonce, CommandId("saved-command"), 0, "/tmp", "saved"), + ) + withTimeout(5_000) { checkpointStarted.await() } + manager.submitCommand("sleep 30") + session.awaitSent(3) + session.disconnects.emit(Unit) + withTimeout(5_000) { manager.state.first { it is SessionState.Failed } } + delay(100) + assertTrue(archives.isEmpty()) + + releaseCheckpoint.complete(Unit) + withTimeout(5_000) { while (archives.size < 2) delay(10) } + assertEquals(archives.first().id, archives.last().id) + assertEquals(1, archives.first().transcript.turns.size) + assertEquals(2, archives.last().transcript.turns.size) + assertEquals(CommandStatus.DISCONNECTED, archives.last().transcript.turns.last().status) + } finally { + releaseCheckpoint.complete(Unit) + manager.disconnect() + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } + } + } + @Test fun `ephemeral session never reaches transcript archive sink`() = runBlocking { val session = RecordingSession() @@ -630,16 +726,18 @@ class SessionManagerIoTest { manager.prepareConnection(fixtureRequest(ephemeral = true)) manager.connectPrepared() - withTimeout(2_000) { + withTimeout(5_000) { manager.state.filterIsInstance().first() } + session.awaitSent(1) session.output.send( lifecycleBytes(nonce, CommandId("bootstrap-probe"), 0, "/tmp"), ) - withTimeout(2_000) { + withTimeout(5_000) { manager.structuredState.filterIsInstance().first() } manager.submitCommand("printf private") + session.awaitSent(2) session.output.send( lifecycleBytes( nonce, @@ -649,14 +747,15 @@ class SessionManagerIoTest { "private output", ), ) - withTimeout(2_000) { + withTimeout(5_000) { manager.transcriptState.first { it.turns.singleOrNull()?.status == CommandStatus.SUCCEEDED } } + assertTrue(archiveSink.archives.isEmpty()) manager.disconnect() - withTimeout(2_000) { + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } @@ -683,25 +782,29 @@ class SessionManagerIoTest { manager.prepareConnection(fixtureRequest()) manager.connectPrepared() - withTimeout(2_000) { + withTimeout(5_000) { manager.state.filterIsInstance().first() } + session.awaitSent(1) session.output.send(lifecycleBytes(nonce, CommandId("bootstrap-probe"), 0, "/tmp")) - withTimeout(2_000) { + withTimeout(5_000) { manager.structuredState.filterIsInstance().first() } manager.submitCommand("printf private") + session.awaitSent(2) session.output.send( lifecycleBytes(nonce, CommandId("saved-command"), 0, "/tmp", "private output"), ) - withTimeout(2_000) { + withTimeout(5_000) { manager.transcriptState.first { it.turns.singleOrNull()?.status == CommandStatus.SUCCEEDED } } + withTimeout(5_000) { manager.transcriptSaveFailed.first { it } } + assertTrue(manager.state.value is SessionState.Connected) manager.disconnect() - withTimeout(2_000) { + withTimeout(5_000) { manager.state.first { it is SessionState.Disconnected } } @@ -895,7 +998,7 @@ private fun SessionCredential.isCleared(): Boolean = when (this) { private class RecordingSession : LiveSshSession { override val output = Channel(Channel.BUFFERED) - override val disconnects: Flow = MutableSharedFlow() + override val disconnects = MutableSharedFlow() val sent = CopyOnWriteArrayList() val disconnected = CompletableDeferred() @@ -904,6 +1007,10 @@ private class RecordingSession : LiveSshSession { sent += bytes.copyOf() } + suspend fun awaitSent(count: Int) { + withTimeout(5_000) { while (sent.size < count) delay(10) } + } + override suspend fun resize(size: TerminalSize): Boolean = true override suspend fun disconnect() { diff --git a/fixtures/openssh/test-android-transcript-process-death.sh b/fixtures/openssh/test-android-transcript-process-death.sh new file mode 100755 index 0000000..4f70057 --- /dev/null +++ b/fixtures/openssh/test-android-transcript-process-death.sh @@ -0,0 +1,97 @@ +#!/bin/sh +set -eu + +script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd) +repository_dir=$(CDPATH= cd -- "$script_dir/../.." && pwd) +THREADLINE_METADATA_REPOSITORY_DIR=$repository_dir +. "$repository_dir/scripts/project-metadata.sh" +unset THREADLINE_METADATA_REPOSITORY_DIR + +adb_command=${ADB:-${ANDROID_HOME:-$HOME/Android/Sdk}/platform-tools/adb} +if [ ! -x "$adb_command" ]; then + echo "Set ADB or ANDROID_HOME to an Android SDK with adb." >&2 + exit 1 +fi +if [ "$("$adb_command" shell getprop ro.kernel.qemu | tr -d '\r')" != 1 ]; then + echo "Run this fixture test on the Android emulator." >&2 + exit 1 +fi + +cd "$script_dir" +host_fingerprint=$(docker compose exec -T openssh \ + ssh-keygen -lf /var/lib/threadline-ssh/ssh_host_ed25519_key.pub | awk '{print $2}') + +cd "$repository_dir" +./gradlew assembleDebug assembleDebugAndroidTest +"$adb_command" install -r app/build/outputs/apk/debug/app-debug.apk >/dev/null +"$adb_command" install -r app/build/outputs/apk/androidTest/debug/app-debug-androidTest.apk \ + >/dev/null + +fixture_key_file=files/threadline-fixture-private-key +preparation_output=$(mktemp) +preparation_pid= +cleanup() { + if [ -n "$preparation_pid" ]; then + "$adb_command" shell am force-stop "$THREADLINE_DEBUG_APPLICATION_ID" >/dev/null 2>&1 || true + wait "$preparation_pid" 2>/dev/null || true + fi + "$adb_command" shell run-as "$THREADLINE_DEBUG_APPLICATION_ID" rm -f "$fixture_key_file" \ + >/dev/null 2>&1 || true + rm -f "$preparation_output" +} +trap cleanup EXIT INT TERM + +test_class=dev.threadline.core.session.AndroidTranscriptCheckpointProcessDeathTest +test_runner=$THREADLINE_DEBUG_TEST_APPLICATION_ID/androidx.test.runner.AndroidJUnitRunner +for ephemeral in false true; do + "$adb_command" shell run-as "$THREADLINE_DEBUG_APPLICATION_ID" mkdir -p files + "$adb_command" exec-in run-as "$THREADLINE_DEBUG_APPLICATION_ID" sh -c \ + "cat > $fixture_key_file" < "$script_dir/.state/client_ed25519" + + : > "$preparation_output" + "$adb_command" shell am instrument -w \ + -e class "$test_class" \ + -e threadlineCheckpointPhase prepare \ + -e threadlineEphemeral "$ephemeral" \ + -e threadlineFixtureHostKeyFingerprint "$host_fingerprint" \ + "$test_runner" > "$preparation_output" 2>&1 & + preparation_pid=$! + + ready=false + attempts=0 + while [ "$attempts" -lt 60 ]; do + case "$(cat "$preparation_output")" in + *THREADLINE_CHECKPOINT_READY*) ready=true; break ;; + esac + if ! kill -0 "$preparation_pid" 2>/dev/null; then break; fi + attempts=$((attempts + 1)) + sleep 1 + done + if [ "$ready" != true ]; then + cat "$preparation_output" + echo "Process-death preparation did not reach its live checkpoint." >&2 + exit 1 + fi + app_pid=$("$adb_command" shell pidof "$THREADLINE_DEBUG_APPLICATION_ID") + [ -n "$app_pid" ] + "$adb_command" shell am force-stop "$THREADLINE_DEBUG_APPLICATION_ID" + wait "$preparation_pid" || true + preparation_pid= + if "$adb_command" shell pidof "$THREADLINE_DEBUG_APPLICATION_ID" >/dev/null; then + echo "The preparing application process is still alive." >&2 + exit 1 + fi + echo "Killed live app process $app_pid (ephemeral=$ephemeral)." + + verification_output=$("$adb_command" shell am instrument -w \ + -e class "$test_class" \ + -e threadlineCheckpointPhase verify \ + -e threadlineEphemeral "$ephemeral" \ + "$test_runner") + printf '%s\n' "$verification_output" + case "$verification_output" in + *"FAILURES!!!"*|*"INSTRUMENTATION_FAILED:"*) exit 1 ;; + *"OK ("*) ;; + *) echo "Process-death verification did not report success." >&2; exit 1 ;; + esac +done