converter) {
return new FileObjectQueue<>(qf, converter);
}
@@ -54,6 +59,12 @@ public boolean isEmpty() {
/** Enqueues an entry that can be processed at any time. */
public abstract void add(T entry) throws IOException;
+ /**
+ * Flushes any buffered writes to storage. No-op for queues that already write synchronously or
+ * are purely in-memory.
+ */
+ public void sync() throws IOException {}
+
/**
* Returns the head of the queue, or {@code null} if the queue is empty. Does not modify the
* queue.
diff --git a/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java b/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java
index bc2ed568267..137f1cd9156 100644
--- a/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java
+++ b/sentry/src/main/java/io/sentry/cache/tape/QueueFile.java
@@ -1,5 +1,6 @@
/*
* Adapted from: https://github.com/square/tape/tree/445cd3fd0a7b3ec48c9ea3e0e86663fe6d3735d8/tape/src/main/java/com/squareup/tape2
+ * Upstream was archived on 2024-10-25 and receives no further fixes; this copy has local changes.
*
* Copyright (C) 2010 Square, Inc.
*
@@ -33,12 +34,19 @@
import org.jetbrains.annotations.Nullable;
/**
- * A reliable, efficient, file-based, FIFO queue. Additions and removals are O(1). All operations
- * are atomic. Writes are synchronous; data will be written to disk before an operation returns. The
- * underlying file is structured to survive process and even system crashes. If an I/O exception is
- * thrown during a mutating change, the change is aborted. It is safe to continue to use a {@code
+ * A reliable, efficient, file-based, FIFO queue. Additions and removals are O(1). The underlying
+ * file is structured to survive process and even system crashes. If an I/O exception is thrown
+ * during a mutating change, the change is aborted. It is safe to continue to use a {@code
* QueueFile} instance after an exception.
*
+ * By default all operations are atomic and writes are synchronous: data is on disk before an
+ * operation returns. Queues built with {@link Builder#synchronousWrites(boolean)
+ * synchronousWrites(false)} leave writes in the OS page cache until {@link #sync()} is called,
+ * which lets callers batch many mutations behind a single fsync. Such writes still survive process
+ * death and are immediately visible to readers, but a system crash before the next {@code sync()}
+ * can lose recent mutations — and because the OS may flush the data and header pages in either
+ * order, it can leave a committed header pointing at data that never reached disk.
+ *
*
Note that this implementation is not synchronized.
*
*
In a traditional queue, the remove operation returns an element. In this queue, {@link #peek}
@@ -50,10 +58,17 @@
*
NOTE: The current implementation is built for file systems that support
* atomic segment writes (like YAFFS). Most conventional file systems don't support this; if the
* power goes out while writing a segment, the segment will contain garbage and the file will be
- * corrupt. We'll add journaling support so this class can be used with more file systems later.
+ * corrupt. Upstream intended to add journaling support for those file systems but never did; this
+ * fork instead accepts the risk and recovers by deleting and recreating the file, losing its
+ * contents (see {@code ringRead}).
*
*
Construct instances with {@link Builder}.
*
+ *
This is a vendored fork. Square archived Tape on 2024-10-25, so the upstream
+ * link in the header is a historical reference, not a source of fixes. This copy has diverged:
+ * corruption recovery, a bounded ring size, and optional buffered writes. Report problems against
+ * sentry-java rather than upstream.
+ *
* @author Bob Lee (bob@squareup.com)
*/
@ApiStatus.Internal
@@ -130,13 +145,22 @@ public final class QueueFile implements Closeable, Iterable {
/** A number of elements at which this queue will wrap around (ring buffer). */
private final int maxElements;
+ /**
+ * When {@code false}, writes are buffered by the OS and callers must call {@link #sync()} to
+ * write to disk. This lets callers batch many adds behind a single fsync instead of paying one
+ * fsync per write.
+ */
+ private final boolean synchronousWrites;
+
boolean closed;
- static RandomAccessFile initializeFromFile(File file) throws IOException {
+ static RandomAccessFile initializeFromFile(File file, boolean synchronousWrites)
+ throws IOException {
if (!file.exists()) {
// Use a temp file so we don't leave a partially-initialized file.
File tempFile = new File(file.getPath() + ".tmp");
- RandomAccessFile raf = open(tempFile);
+ // The one-time initialization is always synchronous so the header is written before rename.
+ RandomAccessFile raf = open(tempFile, true);
try {
raf.setLength(INITIAL_LENGTH);
raf.seek(0);
@@ -152,23 +176,36 @@ static RandomAccessFile initializeFromFile(File file) throws IOException {
}
}
- return open(file);
+ return open(file, synchronousWrites);
}
- /** Opens a random access file that writes synchronously. */
- private static RandomAccessFile open(File file) throws FileNotFoundException {
- return new RandomAccessFile(file, "rwd");
+ private static RandomAccessFile open(File file, boolean synchronousWrites)
+ throws FileNotFoundException {
+ return new RandomAccessFile(file, synchronousWrites ? "rwd" : "rw");
}
- QueueFile(File file, RandomAccessFile raf, boolean zero, int maxElements) throws IOException {
+ QueueFile(
+ File file, RandomAccessFile raf, boolean zero, int maxElements, boolean synchronousWrites)
+ throws IOException {
this.file = file;
this.raf = raf;
this.zero = zero;
this.maxElements = maxElements;
+ this.synchronousWrites = synchronousWrites;
readInitialData();
}
+ /**
+ * Flushes buffered writes to storage. No-op when the queue was opened with synchronous writes,
+ * since those are already durable.
+ */
+ public void sync() throws IOException {
+ if (!synchronousWrites) {
+ raf.getChannel().force(false);
+ }
+ }
+
private void readInitialData() throws IOException {
raf.seek(0);
raf.readFully(buffer);
@@ -196,7 +233,7 @@ private void readInitialData() throws IOException {
private void resetFile() throws IOException {
raf.close();
file.delete();
- raf = initializeFromFile(file);
+ raf = initializeFromFile(file, synchronousWrites);
readInitialData();
}
@@ -767,6 +804,7 @@ public static final class Builder {
final File file;
boolean zero = true;
int size = -1;
+ boolean synchronousWrites = true;
/** Start constructing a new queue backed by the given file. */
public Builder(File file) {
@@ -788,15 +826,24 @@ public Builder size(int size) {
return this;
}
+ /**
+ * When {@code false}, writes are buffered and the caller must call {@link QueueFile#sync()} to
+ * write to disk. Defaults to {@code true} (every write is fsync'd).
+ */
+ public Builder synchronousWrites(boolean synchronousWrites) {
+ this.synchronousWrites = synchronousWrites;
+ return this;
+ }
+
/**
* Constructs a new queue backed by the given builder. Only one instance should access a given
* file at a time.
*/
public QueueFile build() throws IOException {
- RandomAccessFile raf = initializeFromFile(file);
+ RandomAccessFile raf = initializeFromFile(file, synchronousWrites);
QueueFile qf = null;
try {
- qf = new QueueFile(file, raf, zero, size);
+ qf = new QueueFile(file, raf, zero, size, synchronousWrites);
return qf;
} finally {
if (qf == null) {
diff --git a/sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt b/sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt
new file mode 100644
index 00000000000..eec38e151ee
--- /dev/null
+++ b/sentry/src/test/java/io/sentry/cache/PersistingScopeObserverBatchingTest.kt
@@ -0,0 +1,132 @@
+package io.sentry.cache
+
+import com.google.common.truth.Truth.assertThat
+import io.sentry.Breadcrumb
+import io.sentry.ISentryExecutorService
+import io.sentry.ISerializer
+import io.sentry.SentryOptions
+import io.sentry.cache.PersistingScopeObserver.BREADCRUMBS_FILENAME
+import io.sentry.cache.PersistingScopeObserver.TRANSACTION_FILENAME
+import io.sentry.test.DeferredExecutorService
+import java.io.Writer
+import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.test.Test
+import org.junit.Rule
+import org.junit.rules.TemporaryFolder
+
+class PersistingScopeObserverBatchingTest {
+ @get:Rule val tmpDir = TemporaryFolder()
+
+ private val options = SentryOptions()
+
+ private fun getSut(executor: ISentryExecutorService): PersistingScopeObserver {
+ options.executorService = executor
+ options.cacheDirPath = tmpDir.newFolder().absolutePath
+ return PersistingScopeObserver(options)
+ }
+
+ private fun PersistingScopeObserver.readTransaction(): String? =
+ read(options, TRANSACTION_FILENAME, String::class.java)
+
+ @Suppress("UNCHECKED_CAST")
+ private fun PersistingScopeObserver.readBreadcrumbs(): List =
+ read(options, BREADCRUMBS_FILENAME, List::class.java) as List
+
+ @Test
+ fun `defers writes until the flush runs`() {
+ val executor = DeferredExecutorService()
+ val sut = getSut(executor)
+
+ sut.setTransaction("MainActivity")
+ assertThat(sut.readTransaction()).isNull()
+
+ executor.runAll()
+ assertThat(sut.readTransaction()).isEqualTo("MainActivity")
+ }
+
+ @Test
+ fun `coalesces repeated writes keeping only the latest value`() {
+ val executor = DeferredExecutorService()
+ val sut = getSut(executor)
+
+ sut.setTransaction("A")
+ sut.setTransaction("B")
+ sut.setTransaction("C")
+ executor.runAll()
+
+ assertThat(sut.readTransaction()).isEqualTo("C")
+ }
+
+ @Test
+ fun `batches breadcrumbs and persists all of them`() {
+ val executor = DeferredExecutorService()
+ val sut = getSut(executor)
+
+ sut.addBreadcrumb(Breadcrumb().apply { message = "one" })
+ sut.addBreadcrumb(Breadcrumb().apply { message = "two" })
+ assertThat(sut.readBreadcrumbs()).isEmpty()
+
+ executor.runAll()
+ assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("one", "two").inOrder()
+ }
+
+ @Test
+ fun `clearing breadcrumbs drops earlier pending adds but keeps later ones`() {
+ val executor = DeferredExecutorService()
+ val sut = getSut(executor)
+
+ sut.addBreadcrumb(Breadcrumb().apply { message = "dropped" })
+ sut.setBreadcrumbs(emptyList())
+ sut.addBreadcrumb(Breadcrumb().apply { message = "kept" })
+ executor.runAll()
+
+ assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("kept")
+ }
+
+ @Test
+ fun `a clear landing mid-flush does not wipe breadcrumbs added after it`() {
+ val executor = DeferredExecutorService()
+ lateinit var sut: PersistingScopeObserver
+ // fires while the flush is draining, i.e. after the first breadcrumb has been written
+ options.setSerializer(
+ SerializerHook(options.serializer) {
+ sut.setBreadcrumbs(emptyList())
+ sut.addBreadcrumb(Breadcrumb().apply { message = "kept" })
+ }
+ )
+ sut = getSut(executor)
+
+ sut.addBreadcrumb(Breadcrumb().apply { message = "dropped" })
+ executor.runAll()
+ executor.runAll()
+
+ assertThat(sut.readBreadcrumbs().map { it.message }).containsExactly("kept")
+ }
+
+ /** Delegates to [delegate], running [onFirstBreadcrumb] once, mid-serialization. */
+ private class SerializerHook(
+ private val delegate: ISerializer,
+ private val onFirstBreadcrumb: () -> Unit,
+ ) : ISerializer by delegate {
+ private val fired = AtomicBoolean(false)
+
+ override fun serialize(entity: T, writer: Writer) {
+ if (entity is Breadcrumb && fired.compareAndSet(false, true)) {
+ onFirstBreadcrumb()
+ }
+ delegate.serialize(entity, writer)
+ }
+ }
+
+ @Test
+ fun `resetCache keeps pending mutations from the current process`() {
+ val executor = DeferredExecutorService()
+ val sut = getSut(executor)
+
+ sut.setTransaction("SetDuringInit")
+ sut.resetCache()
+ executor.runAll()
+
+ assertThat(sut.readTransaction()).isEqualTo("SetDuringInit")
+ }
+}
diff --git a/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt b/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt
index c909b48a9b3..faa2b2e0ba4 100644
--- a/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt
+++ b/sentry/src/test/java/io/sentry/cache/tape/QueueFileTest.kt
@@ -49,7 +49,8 @@ class QueueFileTest {
@get:Rule val folder = TemporaryFolder()
private lateinit var file: File
- private fun newQueueFile(raf: RandomAccessFile): QueueFile = QueueFile(this.file, raf, true, -1)
+ private fun newQueueFile(raf: RandomAccessFile): QueueFile =
+ QueueFile(this.file, raf, true, -1, true)
private fun newQueueFile(zero: Boolean = true, size: Int = -1): QueueFile =
Builder(file).zero(zero).size(size).build()