Implement batch keys in OSS

[BrunoScheufler]

Description

This pull request introduces a new key option on the function batching configuration. When storing batches, the batch pointer now includes a hashed representation of the evaluated expression. Unlike the previous implementation where one function would have at most one associated active batch, batch keys create many batches per function.

Motivation

Batch keys enable event aggregation or grouping by tenants: Events could be counted for user notifications, or processed together to enable other features.

Type of change (choose one)

• Chore (refactors, upgrades, etc.)
• Bug fix (non-breaking change that fixes an issue)
• Security fix (non-breaking change that fixes a potential vulnerability)
• Docs
• New feature (non-breaking change that adds functionality)
• Breaking change (fix or feature that would cause existing functionality not to work as expected)

Checklist

• I've linked any associated issues to this PR. -> https://linear.app/inngest/issue/INN-3010/implement-batch-keys-in-oss-executor
• I've tested my own changes.

Check our Pull Request Guidelines

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Ignored Deployment

Name	Status	Preview	Comments	Updated (UTC)
ui	⬜️ Ignored (Inspect)	Visit Preview		May 22, 2024 10:48am

[linear]

INN-3010 Implement batch keys in OSS executor

[darwin67]

if it's invalid, maybe just return it as the default fn.ID.String() instead?
hardcoding "<invalid>" here is no different from just using the function ID.

maybe also return an error just in case. I suspect it's not going to be used anytime soon, but hopefully we can surface this kind of invalid expression errors to users better.

[BrunoScheufler]

Updated this to return an error and use the fn identifier as fallback, just to clarify, should we pass this up already or silently ignore the error? I guess it does make sense to warn early and condition users to use valid expressions returning strings, what do you think?

[darwin67]

We don't have a way to pass this error back up to the user atm.
It's the same for all expression related evaluation, and we should bubble the evaluation error up somehow so users are aware.

But regardless we don't want to fail the execution because there's an invalid expression since there are no ways to recover from that and runs are essentially lost at that point.

[darwin67]

Looks great! Nice tests as well.
Have a couple of comments and suggestions.
Only need you to add the CEL expr check in IsValid, others are minor suggestions that are not blocking.

Approving the PR so you can merge it once you made the changes.

[darwin67]

ratelimit hashes the key before returning it. could be a something to copy if it makes sense, and can save you that type assertion above.

inngest/pkg/execution/ratelimit/ratelimit.go

Lines 27 to 51 in 16f8e28

	// RateLimitKey returns the rate limiting key given a function ID, rate limit config,
	// and incoming event data.
	func RateLimitKey(ctx context.Context, id uuid.UUID, c inngest.RateLimit, evt map[string]any) (string, error) {
	if c.Key == nil {
	return id.String(), nil
	}
	eval, err := expressions.NewExpressionEvaluator(ctx, *c.Key)
	if err != nil {
	return "", ErrEvaluatingRateLimitExpression
	}
	res, _, err := eval.Evaluate(ctx, expressions.NewData(map[string]any{"event": evt}))
	if err != nil {
	return "", ErrEvaluatingRateLimitExpression
	}
	// Take a checksum of this data. It doesn't matter if this is a map or a string;
	// as long as we're consistent here.
	return hash(res, id), nil
	}
	func hash(res any, id uuid.UUID) string {
	ui := xxhash.Sum64String(fmt.Sprintf("%v", res))
	sum := strconv.FormatUint(ui, 36)
	return fmt.Sprintf("%s-%s", id, sum)
	}

[darwin67]

I see you're already base64 encoding it, so it might not matter.