Data Pipelines

To ensure consistency, PowerSync relies on Write Checkpoints. These checkpoints ensure that clients have uploaded their own local changes/mutations to the server before applying downloaded data from the server to the local database.

The essential requirement is that the client must get a Write Checkpoint after uploading its last write/mutation. Then, when downloading data from the server, the client checks whether the Write Checkpoint is part of the largest sync checkpoint received from the server (i.e. from the PowerSync Service). If it is, the client applies the server-side state to the local database.

The default Write Checkpoints implementation relies on uploads being acknowledged synchronously, i.e. the change persists in the source database (to which PowerSync is connected) before the uploadData call completes.

Problems occur if the persistence in the source database happens asynchronously. If the client’s upload is meant to mutate the source database (and eventually does), but this is delayed, it will effectively seem as if the client’s uploaded changes were reverted on the server, and then applied again thereafter.

Chained data pipelines are a common example of asynchronous uploads — e.g. data uploads are first written to a different upstream database, or a separate queue for processing, and then finally replicated to the ‘source database’ (to which PowerSync is connected).

For example, consider the following data pipeline:

1. The client makes a change locally and the local database is updated.
2. The client uploads this change to the server.
3. The server resolves the request and writes the change into an intermediate database (not the source database yet).
4. The client thinks the upload is complete (i.e. persisted into the source database). It requests a Write Checkpoint from the PowerSync Service.
5. The PowerSync Service increments the replication HEAD in the source database, and creates a Write Checkpoint for the client. The Write Checkpoint number is returned and recorded in the client.
6. The PowerSync Service replicates past the previous replication HEAD (but the changes are still not present in the source database).
7. It should be fine for the client to apply the state of the server to the local database. But the server state does not include the client’s uploaded changes mentioned in #2. This is the same as if the client’s uploaded changes were rejected (not applied) by the server. This results in the client reverting the changes in its local database.
8. Eventually the change is written to the source database, and increments the replication HEAD.
9. The PowerSync Service replicates this change and sends it to the client. The client then reapplies the changes to its local database.

In the above case, the client may see the Write Checkpoint before the data has been replicated. This will cause the client to revert its changes, then apply them again later when it has actually replicated, causing data to “flicker” in the app.

For these use cases, Custom Write Checkpoints should be implemented.

​

Custom Write Checkpoints

Custom Write Checkpoints allow the developer to define Write Checkpoints and insert them into the replication stream directly, instead of relying on the PowerSync Service to create and return them. An example of this is having the backend persist Write Checkpoints to a dedicated table which is processed as part of the replication stream.

The PowerSync Service then needs to process the (ordered) replication events and correlate the checkpoint table changes to Write Checkpoint events.

​

Example Implementation

A self-hosted Node.js demo with Postgres is available here:

​

Implementation Details

This outlines what a Custom Write Checkpoints implementation entails.

​

Custom Write Checkpoint Table

Create a dedicated checkpoints table, which should contain the following checkpoint payload information in some form:

export type CheckpointPayload = {
    /**
     * The user account id
     */
    user_id: string;
    /**
     * The client id relating to the user account.
     * A single user can have multiple clients.
     * A client is analogous to a device session.
     * Checkpoints are tracked separately for each `user_id` + `client_id`.
     */
    client_id: string;
    /**
     * A strictly increasing Write Checkpoint identifier.
     * This number is generated by the application backend.
     */
    checkpoint: bigint;
}

​

Replication Requirements

Replication events for the Custom Write Checkpoint table (checkpoints in this example) need to enabled.

For Postgres, this involves adding the table to the PowerSync logical replication publication, for example:

create publication powersync for table public.lists, public.todos, public.checkpoints;

​

Sync Rules Requirements

You need to enable the write_checkpoints sync event in your Sync Rules configuration. This event should map the rows from the checkpoints table to the CheckpointPayload payload.

# sync-rules.yaml

# Register the custom write_checkpoints event
event_definitions:
  write_checkpoints:
    payloads:
      #  This defines where the replicated Custom Write Checkpoints should be extracted from
      - SELECT user_id, checkpoint, client_id FROM checkpoints

# Define Sync Rules as usual
bucket_definitions:
  global:
    data:
    ...

​

Application

Your application should handle Custom Write Checkpoints on both the frontend and backend.

​

Frontend

Your client backend connector should make a call to the application backend to create a Custom Write Checkpoint record after uploading items in the uploadData method. The Write Checkpoint number should be supplied to the CRUD transactions’ complete method.

 async function uploadData(database: AbstractPowerSyncDatabase): Promise<void> {
    const transaction = await database.getNextCrudTransaction();
    // Get the unique client ID from the PowerSync Database SQLite storage
    const clientId = await db.getClientId();

      for (const operation of transaction.crud) {
        // Upload the items to application backend
        // ....
      }

      await transaction.complete(await getCheckpoint(clientId));
 }

 async function getCheckpoint(clientId: string): string {
  /**
   * Should perform a request to the application backend which should create the 
   * Write Checkpoint record and return the corresponding checkpoint number.
   */
  return "the Write Checkpoint number from the request";
 }

​

Backend

The backend should create a Write Checkpoint record when the client requests it. The record should automatically increment the Write Checkpoint number for the associated user_id and client_id.

​

Postgres Example

With the following table defined in the database…

CREATE TABLE checkpoints (
    user_id VARCHAR(255),
    client_id VARCHAR(255),
    checkpoint INTEGER,
    PRIMARY KEY (user_id, client_id)
);

…the backend should have a route which creates checkpoints records:

router.put('/checkpoint', async (req, res) => {
  if (!req.body) {
    res.status(400).send({
      message: 'Invalid body provided'
    });
    return;
  }

  const client = await pool.connect();

// These could be obtained from the session
  const { user_id = 'UserID', client_id = '1' } = req.body;

  const response = await client.query(
    `
    INSERT
      INTO
        checkpoints
          (user_id, client_id, checkpoint)
      VALUES
          ($1, $2, '1')
    ON
      CONFLICT (user_id, client_id)
    DO
      UPDATE
        SET checkpoint = checkpoints.checkpoint + 1
    RETURNING checkpoint;
    `,
    [user_id, client_id]
  );
  client.release();

  // Return the Write Checkpoint number
  res.status(200).send({
    checkpoint: response.rows[0].checkpoint
  });
});

An example implementation can be seen in the Node.js backend demo, including examples for MongoDB and MySQL.