Skip to content

#8 Feat: Large transaction handling #8

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

Open
ckritzinger wants to merge 1 commit into
base: feat/write-batches
Choose a base branch
from
Open

#8 Feat: Large transaction handling #8

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
44 changes: 44 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,50 @@ cd frontend && pnpm install && pnpm dev
- PowerSync: http://localhost:8080
- Postgres: localhost:5432

## Huge Transaction Flow

`openapi.yaml` defines a stage/commit/poll flow for transactions too
large to fit in a single `postCrudTransaction` call. The client decides
which option to use by comparing the transaction's op count against a
configured `MAX_TRANSACTION_OPS`. If too long, the transation's operations are split into chunks and staged in batches before the transaction is committed.

```mermaid
sequenceDiagram
participant Client
participant Server

Client->>Client: ops.length > MAX_TRANSACTION_OPS?

alt normal transaction
Client->>Server: POST /api/data<br/>{ crud: [...] }
Server-->>Client: 200 OK<br/>{ status: success | retryable_error | fatal_error }
else huge transaction
Client->>Client: split ops into chunks of size <= MAX_TRANSACTION_OPS

Client->>Server: PUT /api/data/transactions/{id}
Note right of Server: begin staging
Server-->>Client: 201 Created (or 200 if already staging)<br/>{ staged_count }

loop each chunk
Client->>Server: POST /api/data/transactions/{id}/operations<br/>[CrudEntry, ...]
Server-->>Client: 202 Accepted<br/>{ staged_count }
end

Client->>Server: POST /api/data/transactions/{id}/commit
Note right of Server: idempotent, can be retried, returns instantly
Server-->>Client: 202 Accepted<br/>{ status: pending | processing }

par background batch job
Server->>Server: run staged operations as a background transaction
and client polls
loop until terminal status
Client->>Server: GET /api/data/transactions/{id}
Server-->>Client: 200 OK<br/>{ status: processing | success | retryable_error | fatal_error }
end
end
end
```

## Generating Types from OpenAPI Spec

Both the backend and frontend generate TypeScript types from the shared `openapi.yaml` spec.
Expand Down
213 changes: 211 additions & 2 deletions openapi.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ paths:
responses:
"200":
description: >
Transaction result. Always returns 200 — the outcome
Transaction result. Always returns 200 — the outcome
is determined by the status field in the response body.
content:
application/json:
Expand Down Expand Up @@ -76,7 +76,200 @@ paths:
schema:
$ref: "#/components/schemas/MessageResponse"

/api/data/transactions/{transactionId}/operations:
post:
operationId: postTransactionOperations
summary: Stage operations as part of a large transaction
description: >
For transactions too large for a single postCrudTransaction call.
Saves the given operations to the transactionId staging area without processing.
beginTransaction must have been called first.
Can be called repeatedly with the same transactionId to accumulate more operations before committing.
Staged operations that are never committed are discarded after a server-defined TTL.
security:
- bearerAuth: []
parameters:
- $ref: "#/components/parameters/TransactionId"
requestBody:
required: true
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/CrudEntry"
responses:
"202":
description: Operations accepted into the staging area for the specified transaction.
content:
application/json:
schema:
$ref: "#/components/schemas/StagedOperationsResponse"
"401":
description: Missing or invalid token
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"404":
description: >
Unknown transaction. Either beginTransaction was never called for this id, or it has been garbage collected.
The client must not assume its earlier operations calls for this id were preserved and should restart with beginTransaction under a fresh id.
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"409":
description: Transaction has already been committed; staging is closed.
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"500":
description: Unexpected server error
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"

/api/data/transactions/{transactionId}/commit:
post:
operationId: commitTransaction
summary: Commit a staged transaction
description: >
Idempotent, calling commit again on an already-committed transaction returns its current status without reprocessing.
Committing runs the staged operations as a background batch, so this always returns immediately with a pending/processing status.
Poll getTransactionStatus for the outcome.
security:
- bearerAuth: []
parameters:
- $ref: "#/components/parameters/TransactionId"
responses:
"202":
description: Commit accepted. Processing started.
content:
application/json:
schema:
$ref: "#/components/schemas/TransactionResponse"
"401":
description: Missing or invalid token
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"404":
description: Unknown transaction, either nothing was begun/staged or this transaction has been garbage collected.
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"500":
description: Unexpected server error
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"

/api/data/transactions/{transactionId}:
put:
operationId: beginTransaction
summary: Begin staging a large transaction
description: >
Creates the staging area for transactionId, identified by the client-generated UUID in the path.
Operations can only be staged against a transaction that has been explicitly begun, so this must be called before postTransactionOperations can be invoked against a transactionId.
Idempotent while the transaction is still staging, meaning it is safe to retry after a dropped connection and will not create a second transaction or duplicate staged operations.
Returns the current staged_count, so a reconnecting client can tell where it left off, or if a transaction has already been committed, or has been garbage collected.
security:
- bearerAuth: []
parameters:
- $ref: "#/components/parameters/TransactionId"
responses:
"201":
description: Transaction begun.
content:
application/json:
schema:
$ref: "#/components/schemas/StagedOperationsResponse"
"200":
description: Transaction was already staging; returned its current state without changing anything.
content:
application/json:
schema:
$ref: "#/components/schemas/StagedOperationsResponse"
"401":
description: Missing or invalid token
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"409":
description: Transaction has already been committed, so it can no longer be begun again.
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"500":
description: Unexpected server error
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"

get:
operationId: getTransactionStatus
summary: Query the status/result of a committed transaction
description: >
Results are retained for a server-defined TTL after commit completes, then garbage collected.
security:
- bearerAuth: []
parameters:
- $ref: "#/components/parameters/TransactionId"
responses:
"200":
description: Current transaction status.
content:
application/json:
schema:
$ref: "#/components/schemas/TransactionResponse"
"401":
description: Missing or invalid token
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"404":
description: Unknown transaction, either nothing was begun/staged or this transaction has been garbage collected.
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"
"500":
description: Unexpected server error
content:
application/json:
schema:
$ref: "#/components/schemas/MessageResponse"

components:
parameters:
TransactionId:
name: transactionId
in: path
required: true
description: Client-generated UUID identifying a staged/committed transaction.
This must be unique for each transaction.
DO NOT reuse CrudTransaction.transaction_id values across transactions.
schema:
type: string
format: uuid

securitySchemes:
bearerAuth:
type: http
scheme: bearer
bearerFormat: JWT
description: Bearer JWT authenticating the writer to the backend.

schemas:
CrudTransaction:
type: object
Expand Down Expand Up @@ -177,8 +370,13 @@ components:
properties:
status:
type: string
enum: [success, retryable_error, fatal_error, not_attempted]
enum: [pending, processing, success, retryable_error, fatal_error, not_attempted]
description: >
pending: transaction committed, batch job not yet started.
Only returned by the staged-transaction commit/status
endpoints, never by postCrudTransaction.
processing: batch job running. Only returned by the
staged-transaction commit/status endpoints.
success: entire transaction persisted, safe to complete.
retryable_error: transient failure, transaction rolled back,
client should retry.
Expand Down Expand Up @@ -219,6 +417,17 @@ components:
type: string
description: Human-readable error detail.

StagedOperationsResponse:
type: object
required: [transaction_id, staged_count]
properties:
transaction_id:
type: string
format: uuid
staged_count:
type: integer
description: Total number of operations staged for this transaction so far, across all operations calls.

MessageResponse:
type: object
required:
Expand Down