Skip to content

chore(backend): harden sqlite concurrent writes and add stress tests #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
hammond01 wants to merge 1 commit into
base: dev
Choose a base branch
from
Draft

chore(backend): harden sqlite concurrent writes and add stress tests #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
130 changes: 100 additions & 30 deletions backend/src/main/java/io/opspace/backend/repository/SqliteActivityLogRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,14 +3,17 @@
import io.opspace.backend.domain.ActivityAction;
import io.opspace.backend.domain.ActivityLog;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;

import java.util.List;
import java.util.function.Supplier;

@Repository
@ConditionalOnProperty(name = "app.storage.type", havingValue = "sqlite", matchIfMissing = true)
public class SqliteActivityLogRepository implements ActivityLogRepository {
private static final int SQLITE_LOCK_RETRY_MAX = 8;

private static final String INSERT_SQL = """
INSERT INTO activity_logs (
Expand All @@ -26,13 +29,27 @@ INSERT INTO activity_logs (
last_updated_by
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
action = excluded.action,
capture_id = excluded.capture_id,
item_id = excluded.item_id,
metadata = excluded.metadata,
created_at = excluded.created_at,
created_by = excluded.created_by,
last_updated_at = excluded.last_updated_at,
last_updated_by = excluded.last_updated_by
""";
private static final String FIND_ALL_SQL = """
SELECT row_id, id, action, capture_id, item_id, metadata, created_at, created_by, last_updated_at, last_updated_by
FROM activity_logs
ORDER BY created_at DESC
""";
private static final String NEXT_ROW_ID_SQL = "SELECT COALESCE(MAX(row_id), 0) + 1 FROM activity_logs";
private static final String BACKFILL_ROW_ID_SQL = """
UPDATE activity_logs
SET row_id = rowid
WHERE id = ? AND row_id IS NULL
""";
private static final String FIND_ROW_ID_BY_ID_SQL = "SELECT row_id FROM activity_logs WHERE id = ?";

private final JdbcTemplate jdbcTemplate;

Expand All @@ -42,35 +59,47 @@ public SqliteActivityLogRepository(JdbcTemplate jdbcTemplate) {

@Override
public ActivityLog save(ActivityLog activityLog) {
Long rowId = activityLog.rowId() != null
? activityLog.rowId()
: jdbcTemplate.queryForObject(NEXT_ROW_ID_SQL, Long.class);
ActivityLog normalized = new ActivityLog(
rowId,
activityLog.id(),
activityLog.action(),
activityLog.captureId(),
activityLog.itemId(),
activityLog.metadata(),
activityLog.createdAt(),
activityLog.createdBy(),
activityLog.lastUpdatedAt(),
activityLog.lastUpdatedBy()
);
jdbcTemplate.update(
INSERT_SQL,
normalized.rowId(),
normalized.id(),
normalized.action().name(),
normalized.captureId(),
normalized.itemId(),
normalized.metadata(),
normalized.createdAt(),
normalized.createdBy(),
normalized.lastUpdatedAt(),
normalized.lastUpdatedBy()
);
return normalized;
return withSqliteLockRetry(() -> {
ActivityLog normalized = new ActivityLog(
activityLog.rowId(),
activityLog.id(),
activityLog.action(),
activityLog.captureId(),
activityLog.itemId(),
activityLog.metadata(),
activityLog.createdAt(),
activityLog.createdBy(),
activityLog.lastUpdatedAt(),
activityLog.lastUpdatedBy()
);
jdbcTemplate.update(
INSERT_SQL,
normalized.rowId(),
normalized.id(),
normalized.action().name(),
normalized.captureId(),
normalized.itemId(),
normalized.metadata(),
normalized.createdAt(),
normalized.createdBy(),
normalized.lastUpdatedAt(),
normalized.lastUpdatedBy()
);
jdbcTemplate.update(BACKFILL_ROW_ID_SQL, normalized.id());
Long persistedRowId = jdbcTemplate.queryForObject(FIND_ROW_ID_BY_ID_SQL, Long.class, normalized.id());
return new ActivityLog(
persistedRowId,
normalized.id(),
normalized.action(),
normalized.captureId(),
normalized.itemId(),
normalized.metadata(),
normalized.createdAt(),
normalized.createdBy(),
normalized.lastUpdatedAt(),
normalized.lastUpdatedBy()
);
});
}

@Override
Expand All @@ -88,4 +117,45 @@ public List<ActivityLog> findAllNewestFirst() {
rs.getString("last_updated_by")
));
}

private <T> T withSqliteLockRetry(Supplier<T> supplier) {
DataAccessException lastException = null;
for (int attempt = 0; attempt < SQLITE_LOCK_RETRY_MAX; attempt++) {
try {
return supplier.get();
} catch (DataAccessException exception) {
lastException = exception;
if (!isSqliteLocked(exception) || attempt == SQLITE_LOCK_RETRY_MAX - 1) {
throw exception;
}
sleepBackoff(attempt);
}
}
throw lastException;
}

private boolean isSqliteLocked(Throwable throwable) {
Throwable current = throwable;
while (current != null) {
String message = current.getMessage();
String className = current.getClass().getName();
boolean sqliteCause = className.contains("SQLite");
if (sqliteCause && message != null
&& (message.contains("SQLITE_LOCKED")
|| message.contains("SQLITE_BUSY")
|| message.contains("database is locked"))) {
return true;
}
current = current.getCause();
}
return false;
}

private void sleepBackoff(int attempt) {
try {
Thread.sleep((long) (attempt + 1) * 6L);
} catch (InterruptedException interruptedException) {
Thread.currentThread().interrupt();
}
}
}
119 changes: 91 additions & 28 deletions backend/src/main/java/io/opspace/backend/repository/SqliteCaptureRepository.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,18 @@

import io.opspace.backend.domain.Capture;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;

import java.util.List;
import java.util.Optional;
import java.util.function.Supplier;

@Repository
@ConditionalOnProperty(name = "app.storage.type", havingValue = "sqlite", matchIfMissing = true)
public class SqliteCaptureRepository implements CaptureRepository {
private static final int SQLITE_LOCK_RETRY_MAX = 8;

private static final String UPSERT_SQL = """
INSERT INTO captures (
Expand Down Expand Up @@ -46,7 +49,12 @@ ON CONFLICT(id) DO UPDATE SET
FROM captures
ORDER BY created_at DESC
""";
private static final String NEXT_ROW_ID_SQL = "SELECT COALESCE(MAX(row_id), 0) + 1 FROM captures";
private static final String BACKFILL_ROW_ID_SQL = """
UPDATE captures
SET row_id = rowid
WHERE id = ? AND row_id IS NULL
""";
private static final String FIND_ROW_ID_BY_ID_SQL = "SELECT row_id FROM captures WHERE id = ?";
private static final String DELETE_BY_ID_SQL = "DELETE FROM captures WHERE id = ?";

private final JdbcTemplate jdbcTemplate;
Expand All @@ -57,33 +65,47 @@ public SqliteCaptureRepository(JdbcTemplate jdbcTemplate) {

@Override
public Capture save(Capture capture) {
Long rowId = capture.rowId() != null ? capture.rowId() : jdbcTemplate.queryForObject(NEXT_ROW_ID_SQL, Long.class);
Capture normalizedCapture = new Capture(
rowId,
capture.id(),
capture.content(),
capture.createdAt(),
capture.createdBy(),
capture.lastUpdatedAt(),
capture.lastUpdatedBy(),
capture.convertedAt(),
capture.convertedItemId(),
capture.deletedAt()
);
jdbcTemplate.update(
UPSERT_SQL,
normalizedCapture.rowId(),
normalizedCapture.id(),
normalizedCapture.content(),
normalizedCapture.createdAt(),
normalizedCapture.createdBy(),
normalizedCapture.lastUpdatedAt(),
normalizedCapture.lastUpdatedBy(),
normalizedCapture.convertedAt(),
normalizedCapture.convertedItemId(),
normalizedCapture.deletedAt()
);
return normalizedCapture;
return withSqliteLockRetry(() -> {
Capture normalizedCapture = new Capture(
capture.rowId(),
capture.id(),
capture.content(),
capture.createdAt(),
capture.createdBy(),
capture.lastUpdatedAt(),
capture.lastUpdatedBy(),
capture.convertedAt(),
capture.convertedItemId(),
capture.deletedAt()
);
jdbcTemplate.update(
UPSERT_SQL,
normalizedCapture.rowId(),
normalizedCapture.id(),
normalizedCapture.content(),
normalizedCapture.createdAt(),
normalizedCapture.createdBy(),
normalizedCapture.lastUpdatedAt(),
normalizedCapture.lastUpdatedBy(),
normalizedCapture.convertedAt(),
normalizedCapture.convertedItemId(),
normalizedCapture.deletedAt()
);
jdbcTemplate.update(BACKFILL_ROW_ID_SQL, normalizedCapture.id());
Long persistedRowId = jdbcTemplate.queryForObject(FIND_ROW_ID_BY_ID_SQL, Long.class, normalizedCapture.id());
return new Capture(
persistedRowId,
normalizedCapture.id(),
normalizedCapture.content(),
normalizedCapture.createdAt(),
normalizedCapture.createdBy(),
normalizedCapture.lastUpdatedAt(),
normalizedCapture.lastUpdatedBy(),
normalizedCapture.convertedAt(),
normalizedCapture.convertedItemId(),
normalizedCapture.deletedAt()
);
});
}

@Override
Expand Down Expand Up @@ -123,4 +145,45 @@ public List<Capture> findAllNewestFirst() {
public void deleteById(String id) {
jdbcTemplate.update(DELETE_BY_ID_SQL, id);
}

private <T> T withSqliteLockRetry(Supplier<T> supplier) {
DataAccessException lastException = null;
for (int attempt = 0; attempt < SQLITE_LOCK_RETRY_MAX; attempt++) {
try {
return supplier.get();
} catch (DataAccessException exception) {
lastException = exception;
if (!isSqliteLocked(exception) || attempt == SQLITE_LOCK_RETRY_MAX - 1) {
throw exception;
}
sleepBackoff(attempt);
}
}
throw lastException;
}

private boolean isSqliteLocked(Throwable throwable) {
Throwable current = throwable;
while (current != null) {
String message = current.getMessage();
String className = current.getClass().getName();
boolean sqliteCause = className.contains("SQLite");
if (sqliteCause && message != null
&& (message.contains("SQLITE_LOCKED")
|| message.contains("SQLITE_BUSY")
|| message.contains("database is locked"))) {
return true;
}
current = current.getCause();
}
return false;
}

private void sleepBackoff(int attempt) {
try {
Thread.sleep((long) (attempt + 1) * 6L);
} catch (InterruptedException interruptedException) {
Thread.currentThread().interrupt();
}
}
}
Loading
Loading