nick/amethyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix: singleton ArtiProxy lifecycle to prevent state file lock conflicts

Browse Source 
The callbackFlow-based TorService created a new ArtiProxy on every flow
collection. When the app paused and resumed, the old instance's file lock
wasn't released before the new one started, causing "Another process has
the lock on our state files" errors.

Redesigned TorService to own a single ArtiProxy with explicit start/stop:
- ArtiProxy is created once and reused across flow re-collections
- State exposed via MutableStateFlow instead of callbackFlow
- Mutex guards start/stop to prevent races
- TorManager calls service.start() and service.stop() explicitly when
  switching between INTERNAL/OFF/EXTERNAL modes
- stop() includes a 2s settle delay for native file lock release

https://claude.ai/code/session_01BApgDd5udqBzMqysSRMpZu
This commit is contained in:
Claude
2026-04-01 12:44:24 +00:00
parent 293fd39a68
commit c83256bac2
2 changed files with 117 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorManager.kt
+4
View File
@@ -58,17 +58,21 @@ class TorManager(
}.transformLatest { (torType, externalSocksPort) -> }.transformLatest { (torType, externalSocksPort) ->
when (torType) { when (torType) {
TorType.INTERNAL -> { TorType.INTERNAL -> {
service.start()
emitAll(service.status) emitAll(service.status)
} }
TorType.OFF -> { TorType.OFF -> {
service.stop()
emit(TorServiceStatus.Off) emit(TorServiceStatus.Off)
} }
TorType.EXTERNAL -> { TorType.EXTERNAL -> {
service.stop()
if (externalSocksPort > 0) { if (externalSocksPort > 0) {
emit(TorServiceStatus.Active(externalSocksPort)) emit(TorServiceStatus.Active(externalSocksPort))
} else { } else {
service.start()
emitAll(service.status) emitAll(service.status)
} }
} }
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/tor/TorService.kt
+113 -120
View File
@@ -25,152 +25,145 @@ import com.vitorpamplona.quartz.utils.Log
import info.guardianproject.arti.ArtiLogListener import info.guardianproject.arti.ArtiLogListener
import info.guardianproject.arti.ArtiProxy import info.guardianproject.arti.ArtiProxy
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.callbackFlow import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.flowOn import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.launch import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.withContext
import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicLong
private const val DEFAULT_SOCKS_PORT = 19050 private const val DEFAULT_SOCKS_PORT = 19050
private const val MAX_PORT_RETRIES = 3 private const val MAX_PORT_RETRIES = 3
private const val BOOTSTRAP_TIMEOUT_MS = 120_000L private const val STOP_SETTLE_MS = 2_000L
private const val BOOTSTRAP_CHECK_INTERVAL_MS = 10_000L
/**
* Manages a single ArtiProxy instance with explicit start/stop lifecycle.
*
* ArtiProxy holds an exclusive lock on state files in the filesystem. Unlike
* the old Android TorService (where bind/unbind was idempotent), we cannot
* create and destroy ArtiProxy instances on every flow collection cycle —
* the file lock from the old instance may not be released before the new
* one tries to acquire it.
*
* Instead, TorService owns a single ArtiProxy and exposes its state via
* a [StateFlow]. TorManager calls [start]/[stop] to control the lifecycle.
*/
class TorService( class TorService(
val context: Context, val context: Context,
) { ) {
val status = private val mutex = Mutex()
callbackFlow { private var artiProxy: ArtiProxy? = null
Log.d("TorService", "Starting Arti Tor Service") private var currentPort: Int = DEFAULT_SOCKS_PORT
trySend(TorServiceStatus.Connecting) private val bootstrapped = AtomicBoolean(false)
private val lastLogTime = AtomicLong(0L)
var socksPort = DEFAULT_SOCKS_PORT private val _status = MutableStateFlow<TorServiceStatus>(TorServiceStatus.Off)
var artiProxy: ArtiProxy? = null val status: StateFlow<TorServiceStatus> = _status.asStateFlow()
val bootstrapped = AtomicBoolean(false)
val lastLogTime = AtomicLong(System.currentTimeMillis())
val logListener = private val logListener =
ArtiLogListener { logLine -> ArtiLogListener { logLine ->
val text = logLine ?: return@ArtiLogListener val text = logLine ?: return@ArtiLogListener
Log.d("TorService") { "Arti: $text" } Log.d("TorService") { "Arti: $text" }
lastLogTime.set(System.currentTimeMillis()) lastLogTime.set(System.currentTimeMillis())
when { when {
text.contains("Sufficiently bootstrapped", ignoreCase = true) || text.contains("Sufficiently bootstrapped", ignoreCase = true) ||
text.contains("is usable", ignoreCase = true) -> { text.contains("is usable", ignoreCase = true) -> {
if (bootstrapped.compareAndSet(false, true)) { if (bootstrapped.compareAndSet(false, true)) {
trySend(TorServiceStatus.Active(socksPort)) _status.value = TorServiceStatus.Active(currentPort)
Log.d("TorService") { "Arti bootstrapped on port $socksPort" } Log.d("TorService") { "Arti bootstrapped on port $currentPort" }
}
}
text.contains("state changed to Stopped", ignoreCase = true) -> {
bootstrapped.set(false)
trySend(TorServiceStatus.Off)
}
text.contains(
"Another process has the lock",
ignoreCase = true,
) -> {
Log.e("TorService") { "Arti state file lock conflict" }
bootstrapped.set(false)
trySend(TorServiceStatus.Off)
}
} }
} }
var lastError: Exception? = null text.contains("state changed to Stopped", ignoreCase = true) -> {
for (attempt in 0 until MAX_PORT_RETRIES) { bootstrapped.set(false)
try { _status.value = TorServiceStatus.Off
artiProxy = }
ArtiProxy
.Builder(context.applicationContext)
.setSocksPort(socksPort)
.setDnsPort(socksPort + 1)
.setLogListener(logListener)
.build()
artiProxy!!.start() text.contains(
lastError = null "Another process has the lock",
break ignoreCase = true,
} catch (e: Exception) { ) -> {
lastError = e Log.e("TorService") { "Arti state file lock conflict" }
Log.e("TorService") { bootstrapped.set(false)
"Failed to start Arti on port $socksPort (attempt ${attempt + 1}): ${e.message}" _status.value = TorServiceStatus.Off
} }
// Stop the failed instance before retrying }
}
suspend fun start() {
mutex.withLock {
if (artiProxy != null) {
// Already running — just ensure status is up to date
if (bootstrapped.get()) {
_status.value = TorServiceStatus.Active(currentPort)
} else {
_status.value = TorServiceStatus.Connecting
}
return
}
_status.value = TorServiceStatus.Connecting
bootstrapped.set(false)
lastLogTime.set(System.currentTimeMillis())
withContext(Dispatchers.IO) {
var socksPort = DEFAULT_SOCKS_PORT
var lastError: Exception? = null
for (attempt in 0 until MAX_PORT_RETRIES) {
try { try {
artiProxy?.stop() val proxy =
} catch (_: Exception) { ArtiProxy
} .Builder(context.applicationContext)
artiProxy = null .setSocksPort(socksPort)
socksPort++ .setDnsPort(socksPort + 1)
} .setLogListener(logListener)
} .build()
if (lastError != null) { proxy.start()
Log.e("TorService") { "Failed to start Arti after $MAX_PORT_RETRIES attempts" } artiProxy = proxy
trySend(TorServiceStatus.Off) currentPort = socksPort
} lastError = null
Log.d("TorService") { "Arti started on port $socksPort" }
// Monitor for bootstrap stalls: if Arti stops producing logs break
// during bootstrap, restart it. } catch (e: Exception) {
val monitorJob = lastError = e
launch { Log.e("TorService") {
val startTime = System.currentTimeMillis() "Failed to start Arti on port $socksPort (attempt ${attempt + 1}): ${e.message}"
while (!bootstrapped.get()) {
delay(BOOTSTRAP_CHECK_INTERVAL_MS)
if (bootstrapped.get()) break
val elapsed = System.currentTimeMillis() - startTime
val timeSinceLastLog = System.currentTimeMillis() - lastLogTime.get()
if (elapsed > BOOTSTRAP_TIMEOUT_MS) {
Log.w("TorService") {
"Arti bootstrap timed out after ${elapsed / 1000}s"
}
// Stop and restart once
try {
artiProxy?.stop()
} catch (_: Exception) {
}
delay(1000)
try {
artiProxy =
ArtiProxy
.Builder(context.applicationContext)
.setSocksPort(socksPort)
.setDnsPort(socksPort + 1)
.setLogListener(logListener)
.build()
artiProxy!!.start()
lastLogTime.set(System.currentTimeMillis())
} catch (e: Exception) {
Log.e("TorService") { "Arti restart failed: ${e.message}" }
trySend(TorServiceStatus.Off)
}
break // Only retry once
} else if (timeSinceLastLog > BOOTSTRAP_CHECK_INTERVAL_MS * 3) {
Log.w("TorService") {
"Arti log stall detected (${timeSinceLastLog / 1000}s since last log)"
}
} }
socksPort++
} }
} }
awaitClose { if (lastError != null) {
Log.d("TorService", "Stopping Arti Tor Service") Log.e("TorService") { "Failed to start Arti after $MAX_PORT_RETRIES attempts" }
monitorJob.cancel() _status.value = TorServiceStatus.Off
}
}
}
}
suspend fun stop() {
mutex.withLock {
val proxy = artiProxy ?: return@withLock
artiProxy = null
bootstrapped.set(false)
Log.d("TorService", "Stopping Arti")
withContext(Dispatchers.IO) {
try { try {
artiProxy?.stop() proxy.stop()
} catch (e: Exception) { } catch (e: Exception) {
Log.d("TorService") { "Failed to stop Arti: ${e.message}" } Log.d("TorService") { "Failed to stop Arti: ${e.message}" }
} }
// Give the native layer time to release file locks
delay(STOP_SETTLE_MS)
} }
}.flowOn(Dispatchers.IO) _status.value = TorServiceStatus.Off
}
}
} }