|
| 1 | +/* |
| 2 | + * Copyright (c) 2026 dexpace and Omar Aljarrah |
| 3 | + * |
| 4 | + * Licensed under the MIT License. See LICENSE in the project root. |
| 5 | + * SPDX-License-Identifier: MIT |
| 6 | + */ |
| 7 | + |
| 8 | +package org.dexpace.sdk.core.lifecycle |
| 9 | + |
| 10 | +import org.dexpace.sdk.core.instrumentation.ClientLogger |
| 11 | +import java.lang.ref.PhantomReference |
| 12 | +import java.lang.ref.ReferenceQueue |
| 13 | +import java.util.Collections |
| 14 | +import java.util.concurrent.atomic.AtomicBoolean |
| 15 | +import java.util.concurrent.locks.ReentrantLock |
| 16 | +import kotlin.concurrent.withLock |
| 17 | + |
| 18 | +/** |
| 19 | + * Opt-in, log-only detector for closeable resources that are never closed. |
| 20 | + * |
| 21 | + * The detector watches resources (response bodies, sources, any [AutoCloseable]) for the |
| 22 | + * pattern where a caller forgets to `close()` and the object is reclaimed by the garbage |
| 23 | + * collector while still "open". When the JVM makes such a resource phantom-reachable, the |
| 24 | + * detector emits a single leak [LeakReport] — by default a `WARN` log line — optionally with |
| 25 | + * the stack trace of where the resource was created. |
| 26 | + * |
| 27 | + * ## What it does **not** do |
| 28 | + * |
| 29 | + * It never closes anything. Auto-closing a caller-owned resource from a GC callback is unsafe |
| 30 | + * (the close could run on an arbitrary thread, at an arbitrary time, against transport state |
| 31 | + * the caller may still expect to own) so the detector is strictly observational. Behaviour of |
| 32 | + * a program is identical whether the detector is on or off; only diagnostics change. |
| 33 | + * |
| 34 | + * ## Off by default |
| 35 | + * |
| 36 | + * [systemDefault] reads the `dexpace.sdk.leakDetection` system property and is **disabled** |
| 37 | + * unless that property is set to `true`. A [Builder] always lets an application enable it |
| 38 | + * explicitly regardless of the property. |
| 39 | + * |
| 40 | + * ## Mechanism (Java 8 compatible) |
| 41 | + * |
| 42 | + * Detection uses a [PhantomReference] per tracked resource plus a [ReferenceQueue] drained by |
| 43 | + * a single daemon thread. This works identically on every JDK from 8 up and needs no |
| 44 | + * `java.lang.ref.Cleaner` (which is 9+). A tracked resource shares an [AtomicBoolean] with its |
| 45 | + * [LeakTracker]; [LeakTracker.closed] flips the flag and de-registers the phantom reference so |
| 46 | + * a cleanly-closed resource is never reported. |
| 47 | + * |
| 48 | + * ## Determinism for tests |
| 49 | + * |
| 50 | + * The reaper thread is started lazily and only when [Builder.startReaperThread] is `true` |
| 51 | + * (the default for [systemDefault]). Tests can construct a detector with the thread disabled |
| 52 | + * and call [drainManually] after forcing a GC to get a deterministic, race-free check that |
| 53 | + * registration and detection fire. |
| 54 | + * |
| 55 | + * ## Thread-safety |
| 56 | + * |
| 57 | + * All public operations are thread-safe. |
| 58 | + */ |
| 59 | +public class LeakDetector private constructor( |
| 60 | + private val enabled: Boolean, |
| 61 | + private val captureCreationStack: Boolean, |
| 62 | + private val listener: LeakListener, |
| 63 | + private val threadName: String, |
| 64 | + startReaperThread: Boolean, |
| 65 | +) { |
| 66 | + private val queue = ReferenceQueue<Any>() |
| 67 | + |
| 68 | + /** Strong refs keep each [TrackedRef] alive until it is enqueued (phantoms are not self-rooted). */ |
| 69 | + private val live: MutableSet<TrackedRef> = Collections.synchronizedSet(HashSet()) |
| 70 | + |
| 71 | + private val reaperLock = ReentrantLock() |
| 72 | + private var reaper: Thread? = null |
| 73 | + |
| 74 | + init { |
| 75 | + if (enabled && startReaperThread) { |
| 76 | + startReaper() |
| 77 | + } |
| 78 | + } |
| 79 | + |
| 80 | + /** |
| 81 | + * Begins tracking [resource]. The returned [LeakTracker] must have [LeakTracker.closed] |
| 82 | + * called from the resource's `close()` to mark a clean shutdown; otherwise the resource is |
| 83 | + * reported as a leak once it is reclaimed. |
| 84 | + * |
| 85 | + * When the detector is disabled this returns a shared no-op tracker and registers nothing, |
| 86 | + * so the overhead on the disabled path is a single field read and no allocation of detector |
| 87 | + * state. |
| 88 | + * |
| 89 | + * @param resource the object whose lifecycle is watched. The detector holds **no** strong |
| 90 | + * reference to it, so tracking never keeps the resource alive. |
| 91 | + * @param description a non-blank label used in the leak report, e.g. `"ResponseBody"`. |
| 92 | + */ |
| 93 | + public fun track( |
| 94 | + resource: Any, |
| 95 | + description: String, |
| 96 | + ): LeakTracker { |
| 97 | + if (!enabled) { |
| 98 | + return NoopTracker |
| 99 | + } |
| 100 | + val stack = if (captureCreationStack) Throwable(STACK_PROBE_MESSAGE).stackTrace else null |
| 101 | + val closedFlag = AtomicBoolean(false) |
| 102 | + val ref = TrackedRef(resource, queue, description, stack, closedFlag) |
| 103 | + live.add(ref) |
| 104 | + return Handle(ref, closedFlag) |
| 105 | + } |
| 106 | + |
| 107 | + /** |
| 108 | + * Wraps [closeable] so that the SDK detects when it is reclaimed without `close()` having |
| 109 | + * been called. The returned [AutoCloseable] delegates `close()` to [closeable] and, on the |
| 110 | + * first close, marks the tracker so no leak is reported. Callers use the returned wrapper in |
| 111 | + * place of the original (e.g. hand it to a caller via `try`-with-resources / `use {}`). |
| 112 | + * |
| 113 | + * This is the by-hand integration point a response-body or stream factory would use today: |
| 114 | + * create the underlying closeable, return `detector.trackCloseable(it, "ResponseBody")`, and |
| 115 | + * an unclosed wrapper surfaces as a `WARN` when the JVM reclaims it. Behaviour is unchanged |
| 116 | + * whether the detector is enabled or not — the wrapper only ever observes, never auto-closes. |
| 117 | + * |
| 118 | + * When the detector is disabled this returns [closeable] unchanged (no wrapper allocation). |
| 119 | + * |
| 120 | + * @param closeable the resource to delegate to and watch. |
| 121 | + * @param description a non-blank label used in the leak report. |
| 122 | + */ |
| 123 | + public fun trackCloseable( |
| 124 | + closeable: AutoCloseable, |
| 125 | + description: String, |
| 126 | + ): AutoCloseable { |
| 127 | + if (!enabled) { |
| 128 | + return closeable |
| 129 | + } |
| 130 | + return TrackedCloseable(closeable, track(closeable, description)) |
| 131 | + } |
| 132 | + |
| 133 | + /** |
| 134 | + * Drains every leak that has already been enqueued by the GC and reports each through the |
| 135 | + * configured [LeakListener], returning how many leaks were reported. |
| 136 | + * |
| 137 | + * This is the deterministic entry point for tests: force a GC (e.g. with a poll loop that |
| 138 | + * allocates pressure), then call `drainManually()` to process whatever the collector has |
| 139 | + * enqueued so far. It is safe to call even when the reaper thread is running, though in |
| 140 | + * production the thread normally drains the queue first. |
| 141 | + */ |
| 142 | + public fun drainManually(): Int { |
| 143 | + var reported = 0 |
| 144 | + while (true) { |
| 145 | + val ref = queue.poll() as? TrackedRef ?: break |
| 146 | + if (report(ref)) { |
| 147 | + reported++ |
| 148 | + } |
| 149 | + } |
| 150 | + return reported |
| 151 | + } |
| 152 | + |
| 153 | + /** Reports [ref] if it was not closed; always de-registers it. Returns `true` if reported. */ |
| 154 | + private fun report(ref: TrackedRef): Boolean { |
| 155 | + live.remove(ref) |
| 156 | + ref.clear() |
| 157 | + if (ref.closed.get()) { |
| 158 | + return false |
| 159 | + } |
| 160 | + try { |
| 161 | + listener.onLeak(LeakReport.create(ref.description, ref.creationStack)) |
| 162 | + } catch (t: Throwable) { |
| 163 | + // A listener must never kill the reaper thread; swallow and keep draining. |
| 164 | + REAPER_LOGGER.atWarning() |
| 165 | + .event("leak.listener.error") |
| 166 | + .cause(t) |
| 167 | + .log("leak listener threw") |
| 168 | + } |
| 169 | + return true |
| 170 | + } |
| 171 | + |
| 172 | + private fun startReaper() { |
| 173 | + reaperLock.withLock { |
| 174 | + if (reaper != null) { |
| 175 | + return |
| 176 | + } |
| 177 | + val t = |
| 178 | + Thread({ |
| 179 | + while (true) { |
| 180 | + try { |
| 181 | + val ref = queue.remove() as? TrackedRef ?: continue |
| 182 | + report(ref) |
| 183 | + } catch (_: InterruptedException) { |
| 184 | + Thread.currentThread().interrupt() |
| 185 | + return@Thread |
| 186 | + } |
| 187 | + } |
| 188 | + }, threadName) |
| 189 | + t.isDaemon = true |
| 190 | + reaper = t |
| 191 | + t.start() |
| 192 | + } |
| 193 | + } |
| 194 | + |
| 195 | + /** |
| 196 | + * Builder for a [LeakDetector]. Use this to enable detection programmatically (independent |
| 197 | + * of the `dexpace.sdk.leakDetection` system property) or to plug in a custom listener. |
| 198 | + */ |
| 199 | + public class Builder { |
| 200 | + private var enabled: Boolean = false |
| 201 | + private var captureCreationStack: Boolean = false |
| 202 | + private var listener: LeakListener = loggingListener() |
| 203 | + private var threadName: String = DEFAULT_THREAD_NAME |
| 204 | + private var startReaperThread: Boolean = true |
| 205 | + |
| 206 | + /** Enables or disables detection. Disabled detectors allocate nothing per [track] call. */ |
| 207 | + public fun enabled(enabled: Boolean): Builder = |
| 208 | + apply { |
| 209 | + this.enabled = enabled |
| 210 | + } |
| 211 | + |
| 212 | + /** |
| 213 | + * When `true`, captures the stack trace at each [track] call and attaches it to the |
| 214 | + * [LeakReport] so leaks can be traced to their creation site. Off by default because |
| 215 | + * stack capture allocates on every tracked resource. |
| 216 | + */ |
| 217 | + public fun captureCreationStack(capture: Boolean): Builder = |
| 218 | + apply { |
| 219 | + this.captureCreationStack = capture |
| 220 | + } |
| 221 | + |
| 222 | + /** Sets the sink for leak reports. Defaults to a SLF4J `WARN` logger. */ |
| 223 | + public fun listener(listener: LeakListener): Builder = |
| 224 | + apply { |
| 225 | + this.listener = listener |
| 226 | + } |
| 227 | + |
| 228 | + /** Overrides the name of the daemon reaper thread. */ |
| 229 | + public fun threadName(name: String): Builder = |
| 230 | + apply { |
| 231 | + this.threadName = name |
| 232 | + } |
| 233 | + |
| 234 | + /** |
| 235 | + * Controls whether the background reaper daemon thread is started. Set to `false` for |
| 236 | + * deterministic tests that drive detection through [drainManually] only. |
| 237 | + */ |
| 238 | + public fun startReaperThread(start: Boolean): Builder = |
| 239 | + apply { |
| 240 | + this.startReaperThread = start |
| 241 | + } |
| 242 | + |
| 243 | + /** Builds the detector. */ |
| 244 | + public fun build(): LeakDetector = |
| 245 | + LeakDetector( |
| 246 | + enabled = enabled, |
| 247 | + captureCreationStack = captureCreationStack, |
| 248 | + listener = listener, |
| 249 | + threadName = threadName, |
| 250 | + startReaperThread = startReaperThread, |
| 251 | + ) |
| 252 | + } |
| 253 | + |
| 254 | + public companion object { |
| 255 | + /** System property that enables the [systemDefault] detector when set to `true`. */ |
| 256 | + public const val ENABLE_PROPERTY: String = "dexpace.sdk.leakDetection" |
| 257 | + |
| 258 | + /** System property that enables creation-stack capture for the [systemDefault] detector. */ |
| 259 | + public const val CAPTURE_STACK_PROPERTY: String = "dexpace.sdk.leakDetection.captureStack" |
| 260 | + |
| 261 | + /** Default name of the background reaper daemon thread. */ |
| 262 | + public const val DEFAULT_THREAD_NAME: String = "dexpace-leak-detector" |
| 263 | + |
| 264 | + private val REAPER_LOGGER = ClientLogger("org.dexpace.sdk.core.lifecycle.LeakDetector") |
| 265 | + |
| 266 | + private const val STACK_PROBE_MESSAGE = "leak-detector creation-stack probe" |
| 267 | + |
| 268 | + /** |
| 269 | + * The process-wide detector configured from system properties. It is enabled only when |
| 270 | + * `-Ddexpace.sdk.leakDetection=true` is set, and captures creation stacks only when |
| 271 | + * `-Ddexpace.sdk.leakDetection.captureStack=true` is also set. Off by default, so the |
| 272 | + * SDK never spends cycles on leak tracking unless an operator opts in. |
| 273 | + */ |
| 274 | + @JvmField |
| 275 | + public val systemDefault: LeakDetector = |
| 276 | + Builder() |
| 277 | + .enabled("true".equals(System.getProperty(ENABLE_PROPERTY), ignoreCase = true)) |
| 278 | + .captureCreationStack("true".equals(System.getProperty(CAPTURE_STACK_PROPERTY), ignoreCase = true)) |
| 279 | + .build() |
| 280 | + |
| 281 | + /** Returns a [LeakListener] that logs each report at SLF4J `WARN`, with creation stack if present. */ |
| 282 | + @JvmStatic |
| 283 | + public fun loggingListener(): LeakListener = |
| 284 | + LeakListener { report -> |
| 285 | + val event = |
| 286 | + REAPER_LOGGER.atWarning() |
| 287 | + .event("resource.leak") |
| 288 | + .field("resource", report.description) |
| 289 | + val stack = report.creationStack |
| 290 | + if (stack != null) { |
| 291 | + event.cause(creationTrace(report.description, stack)) |
| 292 | + } |
| 293 | + event.log(report.summary()) |
| 294 | + } |
| 295 | + |
| 296 | + /** Wraps a captured creation stack in a throwable so loggers render it as a "cause". */ |
| 297 | + private fun creationTrace( |
| 298 | + description: String, |
| 299 | + stack: Array<StackTraceElement>, |
| 300 | + ): Throwable = |
| 301 | + Throwable("creation site of leaked resource: $description").apply { |
| 302 | + stackTrace = stack |
| 303 | + } |
| 304 | + |
| 305 | + private val NoopTracker: LeakTracker = LeakTracker { } |
| 306 | + } |
| 307 | + |
| 308 | + /** |
| 309 | + * Phantom reference to a tracked resource. Carries the metadata needed to report a leak |
| 310 | + * without ever dereferencing the (already-collected) resource. |
| 311 | + */ |
| 312 | + private class TrackedRef( |
| 313 | + referent: Any, |
| 314 | + queue: ReferenceQueue<Any>, |
| 315 | + val description: String, |
| 316 | + val creationStack: Array<StackTraceElement>?, |
| 317 | + val closed: AtomicBoolean, |
| 318 | + ) : PhantomReference<Any>(referent, queue) |
| 319 | + |
| 320 | + /** Delegating [AutoCloseable] that marks its [LeakTracker] closed on the first `close()`. */ |
| 321 | + private class TrackedCloseable( |
| 322 | + private val delegate: AutoCloseable, |
| 323 | + private val tracker: LeakTracker, |
| 324 | + ) : AutoCloseable { |
| 325 | + override fun close() { |
| 326 | + try { |
| 327 | + delegate.close() |
| 328 | + } finally { |
| 329 | + tracker.closed() |
| 330 | + } |
| 331 | + } |
| 332 | + } |
| 333 | + |
| 334 | + /** Caller-facing handle: flips the shared flag and de-registers on [closed]. */ |
| 335 | + private inner class Handle( |
| 336 | + private val ref: TrackedRef, |
| 337 | + private val closedFlag: AtomicBoolean, |
| 338 | + ) : LeakTracker { |
| 339 | + override fun closed() { |
| 340 | + if (closedFlag.compareAndSet(false, true)) { |
| 341 | + live.remove(ref) |
| 342 | + ref.clear() |
| 343 | + } |
| 344 | + } |
| 345 | + } |
| 346 | +} |
0 commit comments