Skip to main content

Advanced: object-store ETL with move-after-read (Ruby)

A bucket trigger polls, and it has no memory of what it has already seen. The only thing that stops an object being delivered again on the next poll is that it is no longer there.

That single fact is what this guide is about. Get it wrong and you build an infinite loop that bills you for every lap.

Source: examples/functions/objectstore-etl-ruby.

What you need​

  • A project, platformctl login, and an S3-compatible bucket you control. The platform does not host the store.
  • About 25 minutes.

after_read has no default, on purpose​

--after-read move | delete | none

Omit it and the trigger is refused:

source.objectstore.after_read must be "move", "delete" or "none" -
there is no default, because a bucket trigger that never retires an
object fires on it forever

That refusal is doing you a favour. The three choices are:

ValueWhat happensWhen it is right
moveThe object is relocated once your handler succeedsYou want an audit trail of what was processed.
deleteThe object is destroyedThe data is already durable elsewhere.
noneIt stays, and fires again on every pollOnly when something else clears the bucket.
The trap underneath the trap

--after-read move needs a destination the trigger is not watching:

after_read: move needs move_to.bucket or move_to.prefix -
moving an object onto itself re-delivers it on every poll

Moving to a prefix inside the watched path is the same infinite loop, arrived at by a route that looks like you did the careful thing. Either move to a different bucket, or watch a prefix and move outside it:

--prefix incoming/ --move-to-prefix processed/

Consume the whole object, or fail​

The platform moves an object away only when your handler succeeded. That turns partial success into data loss: publish some rows, then raise, and the object is left in place — but publish some rows and then return, and the object is moved away with half its rows never delivered.

So the handler parses and validates everything before publishing anything:

handler.rb
# PARSE AND VALIDATE THE WHOLE FILE FIRST.
rows = []
CSV.parse(text, headers: true).each_with_index do |row, i|
  rows << normalise_row(row, i + 2) # +2: one for the header, one for 1-based
end

rows.each_slice(BATCH) do |slice|
  publish(slice.map { |r| { 'text' => JSON.generate(r), 'key' => r['id'] } })
end

A handler that published as it parsed would pass every happy-path test and still lose half a file the first time a row was malformed.

An empty object must succeed​

handler.rb
if text.nil? || text.strip.empty?
  return { 'statusCode' => 200, 'body' => { 'rows' => 0, 'note' => 'empty object' } }
end

Raising on an empty file feels tidy and is a trap: the platform will not retire an object whose handler failed, so an empty file is redelivered on every poll, forever.

Ruby-specific traps​

The status key is a string​

{ 'statusCode' => 200, ... }   # correct
{ statusCode: 200, ... }       # silently wrong

The shim looks for the string 'statusCode'. A symbol key is serialised into the body as an ordinary field and the response defaults to 200 — which looks fine until the day you need a different status.

to_i and to_f never fail​

handler.rb
begin
  cents = (Float(amount) * 100).round
rescue ArgumentError
  raise MalformedObject, "row #{number}: amount #{amount.inspect} is not a number"
end

"N/A".to_f is 0.0. It does not raise, it does not warn — it turns junk into a zero-value order that nobody notices for a month. Float() raises, which is what you want at the edge of a system.

Read both body shapes​

handler.rb
return event['data'] if event['data'].is_a?(String)
raw = event['data_base64']

A non-JSON body arrives as text in data, or base64 in data_base64 when the bytes are not valid UTF-8. Handling only data works until somebody uploads a file with a stray byte in it.

The row number belongs in the message​

"invalid amount" in a 4,000-row file is not actionable. "row 3127: amount "N/A" is not a number" is a fix.

Deploy it​

printf '%s' "$SA_KEY" | platformctl secrets put pipeline-key
platformctl pubsub topics create etl-rows

platformctl functions deploy ./examples/functions/objectstore-etl-ruby --name etl

platformctl secrets bindings set etl PIPELINE_KEY --secret pipeline-key
platformctl secrets bindings apply etl

platformctl serverless triggers create etl-incoming \
  --type objectstore \
  --target etl \
  --bucket my-drops \
  --endpoint https://s3.us-east-1.amazonaws.com \
  --credentials-secret drops-bucket-creds \
  --prefix incoming/ \
  --after-read move \
  --move-to-prefix processed/

Note --prefix incoming/ with --move-to-prefix processed/: the trigger watches one prefix and retires into another, so a processed object is out of the watched path.

Dependencies, if you add any, go in a Gemfile beside handler.rb; Bundler installs them at build time. This example uses only the standard library.

Run it​

printf 'id,email,amount\na1,Ada@Example.COM,10.50\nb2,bob@example.com,3\n' > drop.csv
# upload drop.csv to s3://my-drops/incoming/

Within a minute the object appears under processed/ and two messages are on the topic:

{"id":"a1","email":"ada@example.com","amount_cents":1050}
{"id":"b2","email":"bob@example.com","amount_cents":300}

Note what normalisation did: the email is lowercased and the amount is integer cents, not a float. Money in a float is a rounding bug waiting for a large enough number.

The numbers to compare against​

WhatNumber
Poll interval — time from upload to first byte readup to 60 s
250-row file, warm80-200 ms
Rows per publish call100
Cold start1-2 s

Batching at 100 is a deliberate middle: one call per row is a round trip per row, and one call for the whole file eventually exceeds the request body limit.

Up to 60 seconds is not a bug

The trigger polls. If you upload and check immediately you will conclude it is broken. Wait a minute.

Testing it without a bucket​

Post the CSV straight at the function rather than dropping a file, so you can iterate in seconds instead of a minute:

curl -sX POST "$FUNCTION_URL" -H 'Content-Type: application/json' \
  -d '{"data":"id,email,amount\na1,Ada@Example.COM,10.50\n"}'
Not platformctl invoke

That command posts the agent body shape, {"message": "..."}, so this handler would see no data at all, find an empty object, and correctly report zero rows — a pass that proves nothing. Functions take their body over HTTP.

Traps, at the point you hit them​

TrapWhat you seeWhy
--after-read omittedThe trigger is refused at createNo default, on purpose.
move into the watched prefixThe same object processed foreverThe next poll finds it again. Move outside the watched path.
none chosen by accidentSame, plus a growing billnone is only right when something else clears the bucket.
Publishing as you parseA malformed row loses the rest of the fileThe object is moved away as consumed. Validate everything first.
Raising on an empty fileOne empty object redelivered foreverThe platform retires an object only on success.
:statusCode symbolA different status is ignoredThe shim reads the string key.
.to_f on a money columnZero-value rows nobody noticesto_f does not raise. Use Float().
Only reading dataWorks, then breaks on one fileNon-UTF-8 bytes arrive as data_base64.
Upload, check immediately"Nothing happened"Polling; allow 60 s.

Teardown​

Trigger first — it references the function:

platformctl serverless triggers delete etl-incoming
platformctl delete etl
platformctl pubsub topics delete etl-rows

What it costs to leave running. The function scales to zero and costs no compute while idle. The trigger polls continuously, which is small but not free. The Pub/Sub topic reserves its max_bytes from the project budget the moment it exists, used or not — and consuming messages does not give budget back, so delete the topic or lower its max_bytes. Objects accumulating under processed/ cost whatever your object store charges; the platform does not manage them.

Where next​