Why are these throttled messages executing while earlier message has failed

[mike-hogan]

I have this set of coordinated Inngest functions:

https://github.com/cozemble/breezbook/blob/main/backend/airtable/src/inngest/announceChangesToAirtable.ts

Their purpose is to push mutation events into airtable.

You will notice that I try to limit any messages for a tenant and environment pair to 1, attempting to ensure the correct application order of the events (see https://github.com/cozemble/breezbook/blob/8949f2d5c2819b0597c6c54d9ff1cf3b451543b4/backend/airtable/src/inngest/announceChangesToAirtable.ts#L107)

What I see when I run this locally, is that later events seem to get processed when an earlier event has failed.

What do I have to do to ensure that later events wait if an earlier event has failed?

Thx!

[darwin67]

What do I have to do to ensure that later events wait if an earlier event has failed?

Could you clarify on what you mean by this?
Like you want to pause the dynamically execution for subsequent events if one fails?

If so, what is the criteria for re-enabling the processing?

[mike-hogan]

Hi @darwin67 .

One of the events is a mutation to a customer record. The event "after" that is a creation of a booking record, referencing the customer record. I want the customer event processed to success before processing of the booking event starts. So I want the execution of the virtual queue to block on processing of this failing customer message, as the retry policy does its stuff, and not resume until the event is either dead lettered or succeeds.

Does that help?

[darwin67]

I'm only seeing one function handle-one-change here so I'm not quite clear what you mean by the "event" after.
https://github.com/cozemble/breezbook/blob/0e84cbc78d145ccab91feb5141eae92aab1939f4/backend/airtable/src/inngest/announceChangesToAirtable.ts#L104-L124

I have 2 interpretations of the question you're asking so I'm just going to answer both, and if it's neither, I'd appreciate you provide more context, or help clarify your question further.

The execution need to be in order for a step of procedures but can be ran in parallel if they're not conflicting

Looking at your code, I get the impression that you want to ensure that execution is limited to 1 for the combination of tenantId and environmentId.

Meaning, these need need to be executing in order and one after another.

tenantId, envId
10, 100  <--- 1st
10, 100  <--- 2nd
10, 100  <--- 3rd

However, these can happen in parallel because there is no conflict in terms of uniqueness when you combine tenantId and envId.

tenantId, envId
10, 100
12, 100
12, 101

From your screenshot, I do see a bunch of handle-one-change running, but have you confirmed that they're running in parallel even if when tenantId-environmentId combination is conflicting?

I suspect it's all running because they're because the events' tenantId-environmentId uniqueness is not conflicting. In that case, this is expected behavior.

I don't see handle-one-change has other steps later, or sending out an event to trigger another function so I don't understand what you mean by the "event" after.

Enforce execution one by one regardless of key

We can this internally as singleton execution. Basically there can always be only one execution running at any point in time.
We currently don't support that yet, but we're aware of it.

Does this help?

[mike-hogan]

We're kinda orbiting around each other and coming in on a shared understanding :)

Here is what I have in terms of an event stream that I want to push through handle-one-change:

Tenant ID	Environment ID	Event type	Event body
darwin	prod	upsert.customer	{...event body}
darwin	prod	upsert.booking	{...event body}
darwin	prod	upsert.payment	{...event body}
mike	dev	upsert.customer	{...event body}
mike	dev	upsert.booking	{...event body}
mike	dev	upsert.payment	{...event body}

I am trying to create one processing queue for darwin-prod and another for mike-dev (I'm sorry if I am not using the right Inngest terms here) and both of these can run in parallel, but only one event at a time in each instance.

I was trying to ensure that the upsert.customer for darwin-prod gets processed successfully before the upsert.booking for darwin-prod.

My intuition was that the upsert.customer for darwin-prod would retry until either success or dead letter before processing upsert.booking for darwin-prod. I do not want the upsert.booking to start until upsert.customer succeeds.

And during all this concurrency=1 processing for darwin-prod, the mike-dev stream would be merrily processing away, also one at a time (concurrency=1)

I am confident that Inngest is only running one at a time for a given tenantId-environmentId pair. My problem is that it does not seem to block on a failed event before processing the next. It seems like the retry logic loops independently of event order.

I think I have kinda restated when you said above, and hopefully correctly. Assuming I have, I guess my question is, can Inngest block and retry on a failed event until success or dead letter?

[darwin67]

Thanks for the table, that makes things a lot more clear now.

I think I have kinda restated when you said above, and hopefully correctly. Assuming I have, I guess my question is, can Inngest block and retry on a failed event until success or dead letter?

Answering your question, yes it can.

It seems you're splitting the job into multiple functions, so what I'd recommend would be either put all the steps within a function or to have a function trigger another when it's done with it's job, and not send out the events all at once.
Basically, when you read the code, it should be traceable in order by the eye, just like how you'd expect it to run sequentially.

There are 3 ways to do it with this flow.

upsert.customer -> upsert.booking -> upsert.payment

Function steps

If every logic of this flow has a similar constraint of tenantId + envId with a concurrency of 1, writing one function with multiple steps can do the trick.

here's some pseudo code.

createFunction(
  { id: "handle-customer-flow", concurrency: { /* settings here */ } }.
  { event: "customer/book" }.
  async ({ event, step } => {
    const customer = await step.run("upsert-customer", () => {
       // ... upsert customer logic
       return customer
    })

    const booking = await step.run("book-customer", () => {
      // ...booking logic
      return booking
    })

    // maybe have something in between to verify payment is available
    const success = await step.run("handle-payment", () => {
      // handle payment logic
      return paymentStatus
    })
  
    return { handled: true }
  })
)

This will make sure each step happens in this order and everything here will be applied with the same concurrency constraint.

If a step fails, it'll handle the retry and only proceed if it's successful. if it fails entirely, the whole function will fail, and the subsequent steps won't be ran.

You can always rerun the function with a similar event, so if a failure cause is external, you can basically redo this whole process from start to finish again.

Invoke

Maybe the subsequent steps don't need concurrency constraint, and you only want the start of the flow to be constraint. step.invoke can still mimic the steps approach in a way that can be more efficient.

const bookingHandler = createFunction(
  { id: "book-customer" }, // <=== no constraints here
  { event: "handle-booking" },
  async ({ event, step } => {
    const booking = await step.run("handle the booking", () => {
      // do something here
      return booking
    })

    // maybe you want to do something more here.

    return booking
  })
)

const paymentHandler = createFunction(
  { id: "payment" }, // <=== no constraints here
  { event: "handle-payment" },
  async ({ event, step } => {
    // maybe you want to do something more here.

    const status = await step.run("handle the payment", () => {
      // do something here
      return status
    })

    // maybe you want to do something more here.
    return status
  })
)

const customerUpsert = createFunction(
  { id: "upsert-customer", concurrency: { /* constraints here */ } },
  { event: "upsert-customer" },
  async ({ event, step } => {
    const customer = await step.run("upsert-customer", () => {
      // do something here
      return customer
    })

    const booking = step.invoke("handle booking", {
      function: bookingHandler,
      timeout: "1h"
    })

    const payment = step.invoke("handle payment", {
      function: paymentHandler,
      timeout: "1h"
    })

    return { handled: true }
  })
)

This method is mostly similar to the single function approach, but the subsequent steps have their own functions, so they have more control and flexibility into how complex those functions can be.
step.invoke will make it feel like a function/rpc call but it allows you to keep the core flow within upsert-customer.

The other note about this is, there are no constraints for booking-handler and payment-handler in this case, so they will execute as fast as they can. which likely isn't a problem since they're pretty much guaranteed to be called by upsert-customer, and the uniqueness is most likely guaranteed.

Fanout

The last one if just taking the Invoke approach to a more loose level, using fire and forget.

const bookingHandler = createFunction(
  { id: "book-customer" }, // <=== no constraints here
  { event: "handle-booking" },
  async ({ event, step } => {
    const booking = await step.run("handle the booking", () => {
      // do something here
      return booking
    })

    // maybe you want to do something more here.

    await step.sendEvent({ name: "handle-payment", data: { booking } },)
  })
)

const paymentHandler = createFunction(
  { id: "payment" }, // <=== no constraints here
  { event: "handle-payment" },
  async ({ event, step } => {
    // maybe you want to do something more here.

    const status = await step.run("handle the payment", () => {
      // do something here
      return status
    })

    // maybe you want to do something more here.
    
    return { done: true }
  })
)

const customerUpsert = createFunction(
  { id: "upsert-customer", concurrency: { /* constraints here */ } },
  { event: "upsert-customer" },
  async ({ event, step } => {
    const customer = await step.run("upsert-customer", () => {
      // do something here
      return customer
    })

    await step.sendEvent({ name: "handle-booking", data: { customer } })
  })
)

When each function finishes, it triggers the next one by sending the event, but the main difference here with the Invoke method is upsert-customer will be considered completed the moment it sent out the handle-booking event, and that's also same for handle-booking as well.

This is good when you don't necessarily care what's happening next and just want to finish up the function right away. Typically the fire and forget way of approach.

Recommendation

My recommendation for your use case is either the Function steps or Invoke approach.
I'd imagine you still want to see the whole flow, but depending on how complicated each step can be, they can either be a step.run or be a step.invoke. also you can combine them as much as you like.

But regardless of which one you decide to go with, order of execution will be guaranteed, and it'll make sure the next workflow will only start when the previous one is successful.

[mike-hogan]

This helped a lot Darwin, thanks. It finally sank in this morning, as noted here - https://github.com/cozemble/breezbook/blob/235adf85484711e2be1229d07973edbe7f20bda2/dev-diaries/mikehogan-dev-diary.md?plain=1#L37

[antonio-bluco]

@darwin67 Hey Darwin,

Sorry for jumping on this thread, but I wanted to ask something similar. What if I want to do effectively the same as done above, only for the function as a whole?

Imagine I have a function with 3 steps. This function is called in response to external events. I want to have multiple invocations wait until the first one finished.
Event A -> Triggers Function execution 1
Event B -> Wait until function execution 1 is finished successfully (wait for its retries, etc.) and only proceed when it's finished successfully or fully abandoned.
Event C -> Same, only waiting for the function triggered by event B.

As a concrete use case - let's say you want to send 3 messages in order. You only want to second message 2 when the first has been delivered, and you send message 3 when the 2nd has been delivered. The 'delivered' notification comes from an external event.

So your function is something like

const id = step.run('call-external-api, sendMessageRequest(text));
const addToDb = step.run('add-to-db', db.addMessage(id, text));
step.waitForEvent('triggered-notification', if: 'event.data.id = id')
step.run('mark-as-delivered', db.updateMessage(id, 'delivered'));

And you want to ensure that messages sent to the same user wait for the previous message to be delivered, i.e. wait for the function to finish executing. Basically you want a queue at the 'functions level'.

What would you suggest for this case?