Throttling RFC

[charlypoly]

Request

https://discord.com/channels/1073595399186694185/1073595399958442096/1123572820564250765

Spec proposal

The first beta version could offer a static throttling configuration at the function level:

defer/enrichVerbatim.ts:

import { defer } from "@defer/client"

const enrichVerbatim = async (text: string) => {
	// calling OpenAI...
}

// function-based throttling, with a configuration object
export default defer(enrichVerbatim, {
	concurrency: 10,
  throttling: {
		max: 60,
		minutes: 1
	}
})

// -- OR --

// function-based throttling, with a configuration string
export default defer(enrichVerbatim, {
	concurrency: 10,
  throttling: '60/min'
})

Throttling grouping

Throttling groups are defined once and can be shared between multiple functions, ex: open-ai that apply a 60/min throttling.

defer/index.ts:

import { throttlingGroup } from "@defer/client"

export const OpenAIThrottling = throttlingGroup('open-ai', '60/min')
export const SlackThrottling = throttlingGroup('slack', '20/min')

// -- OR --

export const OpenAIThrottling = throttlingGroup('open-ai', {
	max: 60,
	minutes: 1
})
export const SlackThrottling = throttlingGroup('slack', {
	max: 20,
	minutes: 1
})

defer/enrichVerbatimToSlack.ts:

import { defer } from "@defer/client"
import { OpenAIThrottlingGroup, SlackThrottling } from "./index"

const enrichVerbatimToSlack = async (text: string) => {
	// calling OpenAI...
}

// multi-groups based throttling
export default defer(enrichVerbatim, {
   concurrency: 10,
   throttling: {
      groups: [OpenAIThrottling, SlackThrottling]
   },
})

defer/enrichVerbatim.ts:

import { defer } from "@defer/client"
import { OpenAIThrottlingGroup } from "./index.js"

const summariseArticle = async (text: string) => {
	// calling OpenAI...
}

// group-based throttling
export default defer(summariseArticle, {
   concurrency: 10,
   throttling: {
	   groups: [OpenAIThrottling]
   },
})

---

Please provide any feedback, request, or comment with some pseudo-code examples.
We'll be happy to answer any question!

[marcfalk]

Looks good! Top of mind that would cover our use-cases at the moment 🙌

I prefer:

export const OpenAIThrottling = throttlingGroup('open-ai', {
  max: 60,
  minutes: 1
})

Over:

export const OpenAIThrottling = throttlingGroup('open-ai', '60/min')

Because:

1. I assume it'd work smoother in my types & autocompletion workflow.
2. It is consistent with throttling options for a single function.
3. It allows for more options to be easily added over time.

Example 1

An example of an option - that isn't needed right away but would be a very neat "nice to have" - would be the ability to set a limit for consecutive failed executions for a group. Say I run out of credits at X API, our database or an external API is down, or whatever endpoint we ask during function execution responds with something we didn't expect...

Under the assumption that I'm just gonna "set and forget" background jobs with Defer, and thus fill up a group with +10.000 requests, it would make a lot of sense for me to not waste GB seconds or API credits on executions that are gonna fail anyway.

I assume history/information like this could live at the function level too but I'd use it to make sure that a) we don't spend resources on failed executions related to APIs that are "pay per request", and b) to avoid having to rerun thousands of jobs, which is currently a click-by-click kind of effort in the Defer console? Forcing me to keep a backup job queue too.

I imagine something like:

export const OpenAIThrottling = throttlingGroup('open-ai', {
  max: 60,
  minutes: 1,
  maxConsecutiveRetries: 9
})

Which at 9 consecutive and failed executions would pause executions for that group and send me an alert/email to let me know something's up. I see this more as a function/group setting than I see it as an account setting.

Example 2

Another option that would be needed for some APIs is a way to specify the rate limiting strategy. The RFC currently supports X requests per fixed time window (I assume), essentially allowing X requests at second 59 in minute 1 and X requests at second 1 in minute 2. I've worked with APIs that applied strategies like "leaky bucket" or similar to avoid potential bursts like that. I imagine one would be able to specify an option that influences the distribution of max over minutes. E.g.:

export const OpenAIThrottling = throttlingGroup('open-ai', {
  max: 60,
  minutes: 1,
  strategy: 'fixed-window' | 'sliding-window' | 'leaky-bucket'
})

Specifying that is not super crucial for me at the moment but might be for other customers. However, it is crucial for me to be able to limit executions not by minutes but seconds. I assume I'd be able to do that with a configuration like this?

export const OpenAIThrottling = throttlingGroup('open-ai', {
  max: 5,
  seconds: 1,
})

Lastly

Would probably choose throttle over throttling as it's easier to read and spell but that might just be me ✌️ In general just happy that this is kind of functionality is coming soon 🙏

[charlypoly]

Because:

1. I assume it'd work smoother in my types & autocompletion workflow.
2. It is consistent with throttling options for a single function.
3. It allows for more options to be easily added over time.

Agree! Let's prefer type safety to shorthands ✅

I imagine something like:

export const OpenAIThrottling = throttlingGroup('open-ai', {
  max: 60,
  minutes: 1,
  maxConsecutiveRetries: 9
})

Which at 9 consecutive and failed executions would pause executions for that group and send me an alert/email to let me know something's up. I see this more as a function/group setting than I see it as an account setting.

Definitely an interesting use case!
It would make sense to include a kind of maxConsecutiveRetries later in a "grouped configuration" feature that allows defining templates of background function configurations (including retries, throttling, concurrency, etc).

export const OpenAIThrottling = throttlingGroup('open-ai', {
  max: 60,
  minutes: 1,
  strategy: 'fixed-window' | 'sliding-window' | 'leaky-bucket'
})

Providing some throttling strategies is definitely planned!
Do you have ideas of options available for the 'leaky-bucket' strategy?
It normally requires some notion of points that can be dynamically decreased.

Would probably choose throttle over throttling as it's easier to read and spell but that might just be me ✌️ In general just happy that this is kind of functionality is coming soon 🙏

Noted 💯

[benwoodward]

looks good.

I much prefer the function based throttle configurations. easier to modify.

I'm a bit confused by multi-group throttling—if you assign two throttle groups to a job, will it just pick the throttle config that has a lower number? e.g. it picks the lowest max number and the lowest concurrency number from all assigned groups?

I imagined that jobs would only be part of one group.

another way you could do it is with folders, which might be easier to reason about (this is how I initially imagined it, which seem natural in a SvelteKit context where file based routing is used)

[marcfalk]

I initially had the same confusion as @benwoodward for jobs belonging to multiple groups, and might have made a wrong assumption on how you plan to handle such.

I assume that a job is only run when the rules for X groups are met. And, any job which cannot be run now is prioritized to run at the next possible moment even though a second job scheduled later could also be run at that exact moment. Is that the idea?

Not being able to specify multiple groups could work, I guess, but would limit the number of API providers per worker to one? Hence, force me to split my code by the API providers I use. That might not be ideal if I wanted to split my workers by business processes, which, in our case, contains requests to multiple API providers.

[charlypoly]

@benwoodward @marcfalk, I agree that assigning background functions to multiple groups is unclear and does not bring much value (cc @gearnode: discussed earlier).

another way you could do it is with folders, which might be easier to reason about (this is how I initially imagined it, which seem natural in a SvelteKit context where file based routing is used)

Agree, but we want to keep Defer as much as "framework-agnostic" as possible; we'll always prefer code-based configuration.

--

Let's wait for comments from a few more people before moving to a planning proposal 🚀

[marcfalk]

@benwoodward @marcfalk, I agree that assigning background functions to multiple groups is unclear and does not bring much value (cc @gearnode: discussed earlier).

I certainly see the value. Just think it requires half a page of explanation on how Defer handles a job that is in two groups for which the job meets the limits of group #1 but does not meet the limits of group #2 (at this moment).

I'd very much like to be able to divide my code by business process, and not have to force-split pieces of code that belongs together (from a DX and business point of view) into multiple jobs in order to throttle properly.

I've tried to make some example code to illustrate the difference.

In short I prefer this:

function processCustomerContentRequest(openAiInput) {
  return new Promise(async (resolve, reject) => {
    try {
      // Create
      const record = await db.record.insert({ data: { status: 'PROCESSING' } })

      // Generate
      const openAiData = await fetch('https://api.openai.com/v1/...', {
        method: 'POST',
        body: JSON.stringify({
          data: openAiInput
        })
      }).then(r => r.json())

      // Analyze
      const analysisData = await fetch('https://api.analysis.com/v1/...', {
        method: 'POST',
        body: JSON.stringify({
          data: prepareDataForAnalysis(openAiData)
        })
      }).then(r => r.json())

      // Enrich
      const enrichmentData = await fetch('https://api.enrichment.com/v1/...', {
        method: 'POST',
        body: JSON.stringify({
          data: prepareDataForEnrichment(analysisData)
        })
      }).then(r => r.json())

      // Save
      await db.record.update({
        where: { id: record.id },
        data: {
          status: 'DONE',
          data: formatData(enrichmentData)
        }
      })

      resolve() 
    } catch (e) {
      reject(e)
    }
  })
}
export default defer(processCustomerContentRequest, {
  concurrency: 10,
  throttle: { groups: [openAIGroup, analysisApiGroup, enrichmentApiGroup] }
})

Over this (which comes with more overhead and complexity imo):

Worker 1

function generateData(openAiInput) {
  return new Promise(async (resolve, reject) => {
    try {
      const record = await db.record.insert({ data: { status: 'PROCESSING' } })

      const data = await fetch('https://api.openai.com/v1/...', {
        method: 'POST',
        body: JSON.stringify({ data: openAiInput })
      }).then(r => r.json())

      await db.record.update({
        where: { id: record.id },
        data: { data }
      })

      analyzeData(record.id)

      resolve() 
    } catch (e) {
      reject(e)
    }
  })
}
export default defer(generateData, {
  concurrency: 10,
  throttle: { group: openAiGroup }
})

Worker 2

function analyzeData(id) {
  return new Promise(async (resolve, reject) => {
    try {
      const record = await db.record.findUnique({ where: { id } })

      const data = await fetch('https://api.analysis.com/v1/...', {
        method: 'POST',
        body: JSON.stringify({
          data: prepareDataForAnalysis(record.data)
        })
      }).then(r => r.json())

      await db.record.update({
        where: { id },
        data: { data }
      })

      enrichData(record.id)

      resolve() 
    } catch (e) {
      reject(e)
    }
  })
}
export default defer(analyzeData, {
  concurrency: 10,
  throttle: { group: analysisApiGroup }
})

Worker 3

function enrichData(id) {
  return new Promise(async (resolve, reject) => {
    try {
      const record = await db.record.findUnique({ where: { id } })

      const data = await fetch('https://api.enrichment.com/v1/...', {
        method: 'POST',
        body: JSON.stringify({
          data: prepareDataForEnrichment(record.data)
        })
      }).then(r => r.json())

      await db.record.update({
        where: { id },
        data: {
          status: 'DONE',
          data: formatData(data)
        }
      })

      resolve() 
    } catch (e) {
      reject(e)
    }
  })
}
export default defer(enrichData, {
  concurrency: 10,
  throttle: { group: enrichmentApiGroup }
})

Just wanted to make that clear ✌️ I do recognize that allowing only one group only is way simpler, and I'd be happy to go with that approach for now if it means we get throttling way sooner even though I'd have to refactor a lot of code on our end 😅

EDIT: Typos...

[charlypoly]

I see, cristal clear; thanks!

[marcfalk]

@charlypoly Any update on this feature? ✌️

[charlypoly]

We plan to start the implementation soon; it is part of our current Sprint! ⚡