Batches

Use a batch when one action can't hold the work: importing every row of an uploaded spreadsheet, writing a large export, or walking every record a query finds or every page of a provider's list. A batch runs in the background, a slice at a time. It keeps its place as it goes, so a crash or a deploy carries on where it stopped and nothing is done twice.

The simplest batch walks every record a query finds. This one marks every item in a stocktake as seen, however many there are:

domains/stock/actions/start_stocktake.jadpo:

use stock/types { ItemPage }use stock/actions { mark_seen }use stock/queries { all_items }/// Starts marking every item seen.action start_stocktake() -> BatchRun {    return batches.each(from: all_items(ItemPage {}), each: item -> mark_seen(item))}

batches.each runs mark_seen for each item all_items gives, a page at a time, each page in its own transaction. The batch starts once this action commits, and the BatchRun it returns names it, for batches.status to ask about later.

batches and BatchRun are built into Jadpo. The rest are your app's own, in these files:

domains/stock/actions/mark_seen.jadpo:

use stock/entities { Item }/// Marks one item seen.action mark_seen(item: Item) -> Unit {    var seen: Item = Item.update(item) {        set: {            seen: true        }    }    assert seen.seen}

domains/stock/queries/all_items.jadpo:

use stock/entities { Item }use stock/types { ItemPage }/// Every item, a page at a time, in an order no change to an item moves.query all_items(page: ItemPage) -> Page<Item> {    from: Item    order: [Item.id asc]    page: {        size: page.page_size        after: page.after    }}

domains/stock/types/item_page.jadpo:

/// Where a batch has got to in the list of items.type ItemPage = Input {    page_size: PageSize {        default: 50    }    after: PageCursor {        optional: true    }}

domains/stock/entities/item.jadpo:

/// One item in a stocktake.entity Item {    id: Uuid    seen: Bool    policy: {        create: [Access.system]        read: [Access.system]        update: [Access.system]    }}

PageSize and PageCursor are built in. A batch has no person behind it, so Item lets Access.system in, and a real app lists its people beside it. Run jadpo migrate new initial to make the tables.

Batches over files

You start a batch with a helper: one for each file format and direction, and batches.each for work with no file. Every helper handles small and large inputs alike, so there is one way to do this whatever the size.

domains/catalogue/actions/import_products.jadpo and domains/catalogue/actions/export_products.jadpo:

// domains/catalogue/actions/import_products.jadpoaction import_products(import: ProductImport) -> BatchRun {    return csv.read(        import.sheet,        as: ProductRow,        each: row -> save_product(row, account: import.account_id),        finished: summary -> import_finished(summary, import: import.id),        key: import.id    )}// domains/catalogue/actions/export_products.jadpoaction export_products(request: ProductExport) -> BatchRun {    return csv.write(        exports,        name: "products.csv",        format: CsvFormat.spreadsheet,        from: products_for_export(request.account_id),        each: product -> ProductExportRow {            sku: product.sku            name: product.name            price: product.price        },        finished: done -> send_export_ready(done.file, to: request.requested_by),        key: request.id    )}

Reading a file

A read helper takes the stored File, the row type (as:), an each: that calls one action per row, and a finished: that runs once at the end.

Writing a file

A write helper takes the store to write to, the file's name:, a from: giving the items, an each: that returns the row to write for each item, and a finished:, which receives the stored file.

Walking records or a provider's list: batches.each

batches.each takes a from: and calls an action for each item. from: is a call to a named query returning a Page, which the run walks a page at a time (Order and pages), or a call to a provider's list operation, which it walks by page.next (External services).

each: and finished:

each: and finished: are written in the inline -> form that standard operations may take (Passing functions). item -> save_product(item) calls an action in the same domain for each item, as a subscriber's run: does. When writing a file, each: is instead a pure expression giving the row.

Values they capture from the action around them, such as import.id, are saved with the run so it can carry on later. So only values that can be stored may be captured.

Starting a run, and key:

A helper may be called only in an action. Like emit, it starts only after the calling action commits. It returns a BatchRun at once, and the run goes on in the background. If the action rolls back, the run never starts.

key: makes a repeated start return the same run, so a client that retries doesn't import a file twice. In an app with a tenant, the same key in two tenants starts two runs.

Progress, cancelling and the summary

batches.status(run) gives a BatchProgress: the items done, the estimated total (from the file's size, or the query's count), the problems so far, and the state, running, finished, failed or cancelled.

batches.cancel(run) stops the run after the batch in hand.

finished: receives an ImportSummary for a read or a batches.each, or an ExportSummary for a write. Each holds the count of items done, the count of items with problems and the problems themselves, and for an export, the written File. Each RowProblem holds the row, the column and the rule. These types are described in Values and types.

How a run goes

The run groups its items into batches. Each batch is one transaction that commits its writes and the run's position together, so a crash or a deploy resumes at the next item and nothing is written twice. A file's order is kept, and one run's batches go one after another. Separate runs go side by side. The run's state lives in the store from:'s query reads (How a run is kept).

Each item's action runs inside its batch's transaction, under the rule for a handled failure in All-or-nothing actions. So an item whose action fails is undone on its own and becomes a problem, while the rest carry on.

batches.each is also the one way to make an outside call for each item, since an action can't make outside calls inside for (Loops). When each:'s action makes outside calls, each item is a batch of its own, run as a subscriber's delivery is. Its calls go first, under keys made from the run and the item's position. Then its writes and the run's position commit together. So an item run again replays its calls rather than sending them again.

finished: runs once after the last batch, in a transaction of its own.

Problems never stop a run

A row that can't be decoded, or an item whose action fails, becomes a RowProblem. It holds the row as a spreadsheet numbers it (the header is row 1, the first data row is row 2), the column, the field and the rule, but never the value, so personal data stays out of problem lists and logs.

Only a problem with the whole file stops a run, at the place it occurs or before: a missing required column, a quote that never closes (named by the line where it opened), an encoding the format doesn't declare, or bytes that aren't text.

"All rows or none" isn't a setting. To get it, run the batch twice: once to check the rows without writing, then again to write.

Rows are typed both ways

A row read is an Input, and a row written is an Output. They follow the same rules, problems and JSON encoding as a route's input and output (One input). A column maps to a field by the field's name, or by column: "Email address". Cells written another way, such as dates with the day first, use the wire: encodings (Wire.day_month_year) or a pure decode: function.

A CSV or TSV row holds single values only. A list or an object in one is a compile error that suggests JSON Lines.

Writing CSV names the program that will read it. CsvFormat.spreadsheet writes a byte-order mark and CRLF line endings, and escapes text that looks like a formula. CsvFormat.standard writes CRLF line endings only. A write without format: is a compile error.

Reading takes UTF-8 (a byte-order mark is removed), UTF-16 with a mark, and older encodings the format declares. It never guesses. Writing is UTF-8. A gzip file is read only when you give max_unpacked_size:, and no archives are opened. The format helpers themselves are described in Data formats.

There is no other way

There is no loop over a file's rows, and no way to parse a whole file into a list. store.read(file) returns FileContents, which is for passing bytes on (Files), and the helpers don't accept it. A provider's CSV report is put in a store and read like an upload. json.read given text is only for single JSON documents in memory.

Choose an order no update changes

An export walks live data, so a record changed while it runs appears as its page saw it. To keep every record in exactly once, order from:'s query by a field no update changes, such as id, or a creation time followed by id. A from: ordered by a field that can be updated is type.batch_order_updatable. It is an error you may decline when you know the order is safe (Declining a judgement call).

Batches instead of chained events

Don't page through a large set with an event per page: a chain of events stops at 16 deep (Limits and runtime faults), so it would stop after 16 pages. Use a batch. Work over many pages of a provider's list is a batch too.

A sweep, in full

This sweep marks every item seen, a page at a time, once per sweep, and keeps how many it went through:

type ShelfCode = Text {    min_length: 1    max_length: 40}type ItemPage = Input {    page_size: PageSize {        default: 25    }    after: PageCursor {        optional: true    }}entity Item {    id: Uuid    code: ShelfCode    seen: Bool    policy: {        create: [Access.public]        read: [Access.public, Access.system]        update: [Access.system]    }}entity Stocktake {    id: Uuid    shelf_run: BatchRun?    seen: Int?    policy: {        create: [Access.public]        read: [Access.public, Access.system]        update: [Access.public, Access.system]    }}/// The items, a page at a time, in an order no change to an item moves.query items(page: ItemPage) -> Page<Item> {    from: Item    order: [Item.id asc]    page: {        size: page.page_size        after: page.after    }}/// Marks one item seen.action mark_seen(item: Item) -> Unit {    var seen: Item = Item.update(item) {        set: {            seen: true        }    }    assert seen.seen}/// Keeps how many items the sweep went through.action swept(summary: ImportSummary, stocktake_id: Stocktake.id) fails StocktakeNotFound -> Unit {    attempt Stocktake.update(stocktake_id) {        set: {            seen: summary.items        }    }}/// Starts the sweep over every item, once per sweep.action start_sweep(stocktake_id: Stocktake.id, page: ItemPage) -> BatchRun {    return batches.each(        from: items(page),        each: item -> mark_seen(item),        finished: summary -> swept(summary, stocktake_id),        key: stocktake_id    )}/// How far the sweep has got.action sweep_progress(shelf_run: BatchRun) -> BatchProgress {    return batches.status(shelf_run)}/// Stops the sweep after the page in hand.action stop_sweep(shelf_run: BatchRun) -> Unit {    batches.cancel(shelf_run)}

Details

The exact rules, for when you hit an edge case.