/
Install

Outbox (watch and ack) ​

A producer inserts events. A consumer reads them with watch, a where filter and ack. After a restart it continues from its stored position.

When to use ​

You need a reliable queue next to your data, without Kafka or Redis. Delivery is at least once, so the consumer must be safe to run twice on the same event.

Collection ​

Declared in examples/ndb.yml:

yaml
outbox:
  write: engine
  id: event_id
  time: created_at
  indexes: [topic]
  durability: strict

receipts:
  id: event_id

Produce ​

Each insert is one event. The topic is indexed.

go
topics := []string{"invoice.paid", "user.signup", "invoice.paid", "user.signup", "invoice.paid"}

for i, topic := range topics {
	id := fmt.Sprintf("evt-%d", i+1)

	if _, err := outbox.Insert(ctx, Event{EventID: id, Topic: topic, CreatedAt: time.Now().UTC()}); err != nil {
		return err
	}

	fmt.Printf("produce %s %s\n", id, topic)
}
rb
topics = %w[invoice.paid user.signup invoice.paid user.signup invoice.paid]

topics.each_with_index do |topic, i|
  id = "evt-#{i + 1}"

  ndb.insert("outbox", [{ event_id: id, topic: topic, created_at: Time.now.utc.iso8601(3) }])

  puts "produce #{id} #{topic}"
end
ts
const topics = ['invoice.paid', 'user.signup', 'invoice.paid', 'user.signup', 'invoice.paid']

for (const [i, topic] of topics.entries()) {
  const id = `evt-${i + 1}`

  await outbox.insert({ event_id: id, topic, created_at: new Date() })

  console.log(`produce ${id} ${topic}`)
}
sh
i=0

for topic in invoice.paid user.signup invoice.paid user.signup invoice.paid; do
	i=$((i + 1))

	ndb POST /v1/c/outbox/insert "{\"insert\":[{\"event_id\":\"evt-$i\",\"topic\":\"$topic\",\"created_at\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\"}]}" >/dev/null

	echo "produce evt-$i $topic"
done

The work, safe to repeat ​

The receipt id is the event id. A second delivery fails with duplicate_id, so the work is not done twice.

go
// charge is the work of the consumer. The receipt id is the event id, so a second
// delivery of the same event fails with duplicate_id and the work is not done twice.
func charge(ctx context.Context, receipts *ndb.Coll[Receipt], eventID string) error {
	_, err := receipts.Insert(ctx, Receipt{EventID: eventID, DoneAt: time.Now().UTC()})

	return err
}
rb
# The work of the consumer. The receipt id is the event id, so a second delivery of
# the same event fails with duplicate_id and the work is not done twice.
def charge(ndb, event_id)
  ndb.insert("receipts", [{ event_id: event_id, done_at: Time.now.utc.iso8601(3) }])

  true
rescue Ndb::UniqueViolation
  false
end
ts
// The work of the consumer. The receipt id is the event id, so a second delivery of
// the same event fails with duplicate_id and the work is not done twice.
async function charge(eventId: string): Promise<boolean> {
  try {
    await receipts.insert({ event_id: eventId, done_at: new Date() })

    return true
  } catch (err) {
    if (err instanceof DuplicateIdError) {
      return false
    }

    throw err
  }
}
sh
# The work of the consumer. The receipt id is the event id, so a second delivery of
# the same event fails with duplicate_id (409) and the work is not done twice.
charge() {
	[[ "$(ndb_status POST /v1/c/receipts/insert "{\"insert\":[{\"event_id\":\"$1\",\"done_at\":\"$(date -u +%Y-%m-%dT%H:%M:%SZ)\"}]}")" == "200" ]]
}

Consume, and crash before the ack ​

where leaves out other topics, but the stored position still moves over them. The program acks after the work. For evt-3 it "crashes" after the work and before the ack.

go
paid := ndb.WatchWhere(ndb.Eq("topic", "invoice.paid"))

fmt.Println("consumer billing, run 1 (where topic = invoice.paid):")

for ev, err := range outbox.Watch(ctx, "billing", paid, ndb.Limit(2)) {
	if err != nil {
		return err
	}

	if err := charge(ctx, receipts, ev.ID); err != nil {
		return err
	}

	if ev.ID == "evt-3" {
		fmt.Println("  evt-3 charged, crash before ack")

		break
	}

	if err := outbox.Ack(ctx, ev); err != nil {
		return err
	}

	fmt.Printf("  %s charged, acked\n", ev.ID)
}
rb
paid = { topic: "invoice.paid" }

puts "consumer billing, run 1 (where topic = invoice.paid):"

ndb.watch("outbox", "billing", where: paid, limit: 2) do |ev|
  charge(ndb, ev["id"])

  if ev["id"] == "evt-3"
    puts "  evt-3 charged, crash before ack"

    break
  end

  ndb.ack("outbox", "billing", ev["pos"])

  puts "  #{ev['id']} charged, acked"
end
ts
const paid = { topic: 'invoice.paid' }

console.log('consumer billing, run 1 (where topic = invoice.paid):')

for await (const ev of outbox.watch('billing', { where: paid, limit: 2 })) {
  await charge(ev.id)

  if (ev.id === 'evt-3') {
    console.log('  evt-3 charged, crash before ack')

    break
  }

  await outbox.ack(ev)

  console.log(`  ${ev.id} charged, acked`)
}
sh
watch='{"consumer":"billing","limit":2,"where":{"topic":"invoice.paid"}}'

echo "consumer billing, run 1 (where topic = invoice.paid):"

while read -r pos id; do
	charge "$id" || true

	if [[ "$id" == "evt-3" ]]; then
		echo "  evt-3 charged, crash before ack"

		break
	fi

	ack "$pos"

	echo "  $id charged, acked"
done < <(ndb POST /v1/c/outbox/watch "$watch" | jq -r 'select(.pos) | "\(.pos) \(.id)"')

Restart ​

The same consumer name resumes after the last ack. evt-3 comes again, the receipt shows it was done, and the consumer acks it.

go
fmt.Println("consumer billing, run 2 (restart, resumes after the last ack):")

for ev, err := range outbox.Watch(ctx, "billing", paid, ndb.Limit(2)) {
	if err != nil {
		return err
	}

	err := charge(ctx, receipts, ev.ID)

	switch {
	case errors.Is(err, ndb.ErrDuplicateID):
		fmt.Printf("  %s delivered again, receipt exists, skipped, acked\n", ev.ID)
	case err != nil:
		return err
	default:
		fmt.Printf("  %s charged, acked\n", ev.ID)
	}

	if err := outbox.Ack(ctx, ev); err != nil {
		return err
	}
}
rb
puts "consumer billing, run 2 (restart, resumes after the last ack):"

ndb.watch("outbox", "billing", where: paid, limit: 2) do |ev|
  if charge(ndb, ev["id"])
    puts "  #{ev['id']} charged, acked"
  else
    puts "  #{ev['id']} delivered again, receipt exists, skipped, acked"
  end

  ndb.ack("outbox", "billing", ev["pos"])
end
ts
console.log('consumer billing, run 2 (restart, resumes after the last ack):')

for await (const ev of outbox.watch('billing', { where: paid, limit: 2 })) {
  if (await charge(ev.id)) {
    console.log(`  ${ev.id} charged, acked`)
  } else {
    console.log(`  ${ev.id} delivered again, receipt exists, skipped, acked`)
  }

  await outbox.ack(ev)
}
sh
echo "consumer billing, run 2 (restart, resumes after the last ack):"

while read -r pos id; do
	if charge "$id"; then
		echo "  $id charged, acked"
	else
		echo "  $id delivered again, receipt exists, skipped, acked"
	fi

	ack "$pos"
done < <(ndb POST /v1/c/outbox/watch "$watch" | jq -r 'select(.pos) | "\(.pos) \(.id)"')

Output ​

The four programs print the same text.

text
produce evt-1 invoice.paid
produce evt-2 user.signup
produce evt-3 invoice.paid
produce evt-4 user.signup
produce evt-5 invoice.paid
consumer billing, run 1 (where topic = invoice.paid):
  evt-1 charged, acked
  evt-3 charged, crash before ack
consumer billing, run 2 (restart, resumes after the last ack):
  evt-3 delivered again, receipt exists, skipped, acked
  evt-5 charged, acked
receipts: 3 (one per invoice, none twice)

Notes ​

  • The consumer name (billing) is the stored cursor. Keep it stable.
  • limit: 2 ends the stream after two events, so the example finishes. A real consumer leaves it out and loops.
  • After watch_lagged, call watch again: it resumes after the last ack. history_truncated means wal_retention removed the position. See watch and ack.
  • The Ruby model layer (ndb-rails) has no watch. Use the client, as this example does.
  • Every SDK supports every step of this example.

Run it ​

sh
cd examples
make run-outbox LANG=go   # go, ruby, ts or curl
make run-outbox           # every language

Source: examples/outbox. The programs read NDB_URL and NDB_TOKEN, and they can run again and again.