Eventual Consistency with QueueBear

What is Eventual Consistency?

Eventual consistency is a consistency model where, given enough time without new updates, all replicas of data will converge to the same state. Unlike strong consistency (where all nodes see the same data immediately), eventual consistency allows for temporary inconsistencies while guaranteeing that the system will eventually become consistent.

When to use eventual consistency:

• Distributed microservices that need to share state
• Systems where availability is more important than immediate consistency
• Long-running processes that span multiple services
• Scenarios where network partitions may occur

QueueBear's Delivery Semantics

QueueBear provides at-least-once delivery with deduplication capabilities, which together enable effectively-once semantics:

Message Deduplication

Prevent duplicate message processing at publish-time:

await qb.messages.publish(
  "https://orders.example.com/create",
  { orderId: "order-123", items: [...] },
  {
    deduplicationId: "create-order-123",  // Same ID = same message returned
  }
);

Workflow Idempotency

Ensure workflow runs are unique:

const { runId } = await qb.workflows.trigger(
  "order-fulfillment",
  "https://app.com/workflows/fulfill",
  { orderId: "order-123" },
  {
    idempotencyKey: "fulfill-order-123",  // Same key = same run returned
  }
);

Step Caching: Exactly-Once Step Execution

The most powerful feature for eventual consistency is step caching. When using serve() workflows, each step is executed exactly once, even if the workflow is retried:

export const POST = serve<{ orderId: string }>(async (ctx) => {
  // Step 1: Charge payment
  // If workflow fails after this step and is retried,
  // this step returns the cached result instead of charging again
  const payment = await ctx.run("charge-payment", async () => {
    return await paymentService.charge(ctx.input.orderId, amount);
  });

  // Step 2: Reserve inventory
  // Only executes if step 1 completed successfully
  await ctx.run("reserve-inventory", async () => {
    await inventoryService.reserve(ctx.input.orderId, items);
  });

  return { success: true, paymentId: payment.id };
});

How it works:

1. When a step completes, its result is stored in the database
2. If the workflow is re-executed (after failure, timeout, or sleep), the cached result is returned
3. The step function is NOT called again - preventing duplicate side effects

Consistency Windows

When designing for eventual consistency, understand the timing:

Example: Order System with Eventual Consistency

Here's a complete example of an order processing system:

Workflow Definition

export const POST = serve<OrderInput>(async (ctx) => {
  const { orderId, customerId, items, total } = ctx.input;

  // Step 1: Validate inventory (cached)
  const inventoryCheck = await ctx.run("validate-inventory", async () => {
    const response = await fetch("https://inventory.internal/validate", {
      method: "POST",
      body: JSON.stringify({ items }),
    });
    if (!response.ok) throw new Error("Insufficient inventory");
    return response.json();
  });

  // Step 2: Process payment (cached - critical for consistency)
  const payment = await ctx.run("process-payment", async () => {
    const response = await fetch("https://payments.internal/charge", {
      method: "POST",
      body: JSON.stringify({ customerId, amount: total, orderId }),
    });
    if (!response.ok) throw new Error("Payment failed");
    return response.json();
  });

  // Step 3: Reserve inventory (cached)
  await ctx.run("reserve-inventory", async () => {
    await fetch("https://inventory.internal/reserve", {
      method: "POST",
      body: JSON.stringify({ orderId, items }),
    });
  });

  // Step 4: Create shipment (cached)
  const shipment = await ctx.run("create-shipment", async () => {
    const response = await fetch("https://shipping.internal/create", {
      method: "POST",
      body: JSON.stringify({ orderId, items }),
    });
    return response.json();
  });

  return {
    success: true,
    orderId,
    paymentId: payment.id,
    shipmentId: shipment.id,
  };
});

Triggering with Idempotency

async function processOrder(order: Order) {
  const { runId } = await qb.workflows.trigger(
    "order-fulfillment",
    "https://app.com/api/workflows/order",
    {
      orderId: order.id,
      customerId: order.customerId,
      items: order.items,
      total: order.total,
    },
    {
      // Idempotency key ensures this order is only processed once
      idempotencyKey: `process-order-${order.id}`,
      maxDuration: 3600,  // 1-hour timeout
    }
  );
  return runId;
}

Handling Prolonged Downtime

When a downstream service experiences extended outages:

1. Monitor the Dead Letter Queue

const { entries } = await qb.dlq.list();

for (const entry of entries) {
  console.log(`Failed: ${entry.destination}`);
  console.log(`Reason: ${entry.failureReason}`);
  console.log(`Attempts: ${entry.totalAttempts}`);
}

2. Recover When Service Returns

// Retry all failed messages when service recovers
const results = await qb.dlq.retryAll();
console.log(`Recovered ${results.length} messages`);

3. Use Event-Based Waiting

export const POST = serve<OrderInput>(async (ctx) => {
  await ctx.run("create-order", async () => { /* ... */ });

  // Wait for external service to signal readiness
  const externalResult = await ctx.waitForEvent(
    "wait-for-external-service",
    "external.service.ready",
    {
      eventKey: ctx.input.orderId,
      timeoutSeconds: 86400,  // Wait up to 24 hours
    }
  );

  await ctx.run("complete-with-external", async () => {
    await externalService.complete(externalResult.data);
  });
});

Best Practices

1. Design Idempotent Consumers

app.post("/api/webhooks/order-created", async (req, res) => {
  const { orderId } = req.body;

  // Check if already processed
  const existing = await db.orders.findUnique({ where: { id: orderId } });
  if (existing) {
    return res.json({ success: true, message: "Already processed" });
  }

  await db.orders.create({ data: { id: orderId, ... } });
  res.json({ success: true });
});

2. Use Unique Identifiers

const orderId = crypto.randomUUID();

await qb.messages.publish(endpoint, { orderId }, {
  deduplicationId: `order-${orderId}`,
});

3. Track State Externally

await ctx.run("update-order-status", async () => {
  await db.orders.update({
    where: { id: orderId },
    data: {
      status: "processing",
      lastStepCompleted: "payment",
      updatedAt: new Date(),
    },
  });
});

Summary