Live updates

A user has the chat open on a laptop and a phone. When a reply arrives, both show it at once, without polling. And when a chat grows long, one worker at a time writes its summary, even if several processes try. This chapter uses two engines that keep nothing on the disk: Channels for messages to whoever listens now, and Locks for one holder of a key at a time.

#Open a channel and a lock

interface ChatEvent {
	kind: 'message'
	id: number
}

const events = store.channel<ChatEvent>('chat-events')
const summaries = store.lock('summaries', { ttl: '30s' })

@dataclass
class ChatEvent:
    kind: str
    id: int


events = store.channel("chat-events", ChatEvent)
summaries = store.lock("summaries", ttl="30s")

type ChatEvent struct {
	Kind string `json:"kind"`
	ID   int64  `json:"id"`
}

events, err := channels.OpenChannel[ChatEvent](store, "chat-events")
summaries, err := locks.OpenLock(store, "summaries", locks.TTL(30*time.Second))

#[derive(Serialize, Deserialize, Clone)]
struct ChatEvent {
    kind: String,
    id: i64,
}

let events = store.channel::<ChatEvent>("chat-events").open()?;
let summaries = store.lock("summaries").open()?;

A channel is a name that messages of one type are published on. Every process connected to the store shares it: four web processes and a worker all publish and listen in one place, because they all reach the same store.

#Tell every open tab

The reply worker from Replies in the background saves the reply and publishes an event about it in the same transaction:

await db.tx(async tx => {
	const reply = await tx.exec`insert into messages (chat_id, role, text, created_at, reply_to)
		values (${chatId}, 'assistant', ${answer}, ${Date.now()}, ${messageId})`
	await tx.with(events).under(chatId).publish({ kind: 'message', id: reply.lastInsertRowid })
})

async def save(tx: tinystore.SqlTx) -> None:
    reply = await tx.exec(
        "insert into messages (chat_id, role, text, created_at, reply_to) values (?, 'assistant', ?, ?, ?)",
        chat_id, answer, time.time_ns() // 1_000_000, message_id,
    )
    await tx.with_(events).under(chat_id).publish(ChatEvent(kind="message", id=reply.last_insert_rowid))


await db.tx(save)

err = db.Tx(ctx, func(tx *sqldb.Tx) error {
	reply, err := tx.Exec(ctx, `insert into messages (chat_id, role, text, created_at, reply_to) values (?, 'assistant', ?, ?, ?)`,
		chatID, answer, time.Now().UnixMilli(), messageID)
	if err != nil {
		return err
	}
	return events.In(tx).Under(chatID).Publish(ctx, ChatEvent{Kind: "message", ID: reply.LastInsertRowid})
})

db.tx(|tx| -> tinystore::Result<()> {
    let reply = tx.exec(sql!(
        "insert into messages (chat_id, role, text, created_at, reply_to) values (?, 'assistant', ?, ?, ?)",
        chat_id,
        answer,
        unix_millis(SystemTime::now()),
        message_id
    ))?;
    tx.with(&events).under(chat_id).publish(&ChatEvent { kind: "message".into(), id: reply.last_insert_rowid })?;
    Ok(())
})?;

under(chatId) is a branch of the channel, as a bucket has branches: the messages of one chat. A message published inside a transaction waits for its commit, and a rollback drops it (contract CHAN-03). A tab never hears about a reply that was never saved, and never hears about it before it can read it.

The browser listens through server-sent events. The endpoint subscribes to the chat's branch and forwards each event:

async function stream(chatId: number, send: (data: unknown) => void, closed: Promise<void>) {
	const sub = await events.under(chatId).subscribe(m => send(m.value), {
		onMissed: () => send({ kind: 'reload' }),
	})
	await closed // the browser went away
	await sub.stop()
}

async def stream(chat_id: int, send, closed: asyncio.Event) -> None:
    sub = await events.under(chat_id).subscribe(
        lambda m: send(m.value),
        on_missed=lambda missed: send({"kind": "reload"}),
    )
    await closed.wait()  # the browser went away
    await sub.stop()

func stream(ctx context.Context, chatID int64, send func(any)) error {
	sub, err := events.Under(chatID).Subscribe(ctx, func(ctx context.Context, m channels.Message[ChatEvent]) error {
		send(m.Value)
		return nil
	}, channels.OnMissed(func(channels.Missed) { send(map[string]string{"kind": "reload"}) }))
	if err != nil {
		return err
	}
	<-ctx.Done() // the browser went away
	return sub.Stop(context.Background())
}

fn stream(chat_id: i64, send: impl Fn(TabEvent) + Send + Sync + Clone + 'static) -> tinystore::Result<Subscription> {
    let on_missed = send.clone();
    events
        .under(chat_id)
        .on_missed(move |_| on_missed(TabEvent::Reload))
        .subscribe(move |m| send(TabEvent::Message(m.value)))
}

subscribe returns once the store has the subscription. Every message published after that reaches it, in order (contract CHAN-01). The handler gets one message at a time, never two at once.

#A tab that falls behind

A phone on a slow network cannot take events as fast as a busy chat publishes them. The chat does not wait for it:

const sub = await events.under(chatId).subscribe(m => send(m.value), {
  onMissed: () => send({ kind: 'reload' }),
})

A publisher never waits for a subscriber, and a subscriber that falls behind is told exactly how many messages it missed before its next one (contract CHAN-02). A channel is not a log: the database is. So a missed event costs one reload, never a wrong screen.

The same holds when the connection to the store drops. The SDK subscribes again by itself, and the next call is onMissed with the reason reconnect, because messages may have been lost meanwhile.

#One summary per chat at a time

When a chat passes a hundred messages, the chat asks the model for a summary of it. Two replies finishing at once must not start two summaries of the same chat, so the job holds a lock on the chat's id:

const started = await summaries.tryHold(chatId, async lease => {
	const summary = await model.summarize(await history(chatId))
	const fence = lease.fence.toString().padStart(40, '0')
	await db.exec`update chats set summary = ${summary}, summary_fence = ${fence}
		where id = ${chatId} and summary_fence < ${fence}`
})
if (!started.acquired) return // another worker is writing this chat's summary

async with summaries.try_hold(chat_id) as lease:
    if lease is None:
        return  # another worker is writing this chat's summary
    summary = await model.summarize(await history(chat_id))
    fence = f"{lease.fence:040d}"
    await db.exec(
        "update chats set summary = ?, summary_fence = ? where id = ? and summary_fence < ?",
        summary, fence, chat_id, fence,
    )

started, err := summaries.TryHold(ctx, chatID, func(ctx context.Context, lease *locks.Lease) error {
	summary, err := model.Summarize(ctx, history(ctx, chatID))
	if err != nil {
		return err
	}
	fence := fmt.Sprintf("%040s", lease.Fence)
	_, err = db.Exec(ctx, `update chats set summary = ?, summary_fence = ? where id = ? and summary_fence < ?`,
		summary, fence, chatID, fence)
	return err
})
if err == nil && !started {
	return nil // another worker is writing this chat's summary
}

let started = summaries.try_hold(chat_id, |lease| -> Result<(), AppError> {
    let summary = model.summarize(&history(chat_id)?)?;
    let fence = format!("{:040}", lease.fence().get());
    db.exec(sql!(
        "update chats set summary = ?, summary_fence = ? where id = ? and summary_fence < ?",
        summary,
        &fence,
        chat_id,
        &fence
    ))?;
    Ok(())
})?;
if started.is_none() {
    return Ok(()); // another worker is writing this chat's summary
}

The migration adds the two columns:

alter table chats add column summary text;
alter table chats add column summary_fence text not null default '';

tryHold runs the function only if nobody holds the key, and returns at once otherwise. hold would wait for the key instead, ten seconds unless wait says otherwise. While the function runs, the SDK renews the lease every third of its ttl, and the key is let go when the function returns or throws.

#A worker that stalls

The summary takes a while, and a worker can freeze in the middle: a long garbage-collection pause, a laptop lid closed on a development machine, a virtual machine moved to another host. Its lease runs out, and another worker takes the key. Then the first one wakes up:

const started = await summaries.tryHold(chatId, async lease => {
  const summary = await model.summarize(await history(chatId))
  const fence = lease.fence.toString().padStart(40, '0')
  await db.exec`update chats set summary = ${summary}, summary_fence = ${fence}
    where id = ${chatId} and summary_fence < ${fence}`
})

Every lease the store gives has a fence larger than every earlier one, across restarts too (contract LOCK-02). A lease over a connection lasts its ttl from its last renewal, and one that ran out is never renewed (contract LOCK-05). The fence is what makes this safe: the lock makes two writers at once rare, and the fence makes the late one harmless. Anything that keeps the largest fence it has seen can refuse a late write, as the where clause does here.

The fence is a 128-bit number. Padded to 40 digits, it compares as text in the same order as the number, so SQLite can compare it.

A Rust program runs the lock in the store's own process, where a holder cannot freeze apart from the store, so its lease ends only when its function does. maxHold bounds a lease in every language, for work that must never hold a key longer than a set time.

#Why this design?

Why channels keep nothing on the disk. A typing indicator, "new reply in chat 7", "this tab should reload": these are worth something now and nothing in a minute. Writing them to a file would cost a sync per event and fill the disk with what nobody reads again. What must not be lost belongs in the database or in a queue, and an event only says "look again".

Why a slow subscriber loses messages instead of slowing everyone. The other choice is to make the publisher wait until every subscriber caught up. Then one phone on a train slows down every reply in the chat, and a frozen tab stops the worker. TinyStore keeps a bounded log per branch, lets each subscriber read at its own pace, and tells the slow one what it lost, with an exact count.

Why locks live in memory. One process owns the store, and every holder reaches the store through it, so that process sees every lease and every waiter. Giving and renewing a lease is a change in memory, with no disk sync. After a restart nobody holds anything, which is right: the threads that held the keys are gone. A lease kept on disk would make a restarted store wait out the ttl of holders that no longer exist.

Why a fence and not just a lock. No lock can stop a process that froze while holding it: when it wakes, it does not know that its time ran out. Only the thing it writes to can refuse it, and only if it can tell old from new. The fence gives it that number, as Martin Kleppmann's essay on distributed locks describes. A payment provider that compares nothing still needs an idempotency key: a lock makes double work rare, never impossible.

#Limits and defaults

#See also

• Replies in the background: the worker that publishes the events.
• Channels API reference and Locks API reference: every call of both.