-
Notifications
You must be signed in to change notification settings - Fork 432
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(migrate): implement mutation batcher and use when submitting aga…
…inst mutate endpoint
- Loading branch information
Showing
8 changed files
with
253 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| export const MUTATION_ENDPOINT_MAX_BODY_SIZE = 1024 * 256 // 256KB |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
158 changes: 158 additions & 0 deletions
158
packages/@sanity/migrate/src/runner/utils/__tests__/batchMutations.test.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,158 @@ | ||
| import {at, createIfNotExists, createOrReplace, patch, set} from '@bjoerge/mutiny' | ||
| import {Mutation} from '@sanity/client' | ||
| import {batchMutations} from '../batchMutations' | ||
|
|
||
| function byteLength(obj: unknown) { | ||
| return JSON.stringify(obj).length | ||
| } | ||
|
|
||
| describe('mutation payload batching', () => { | ||
| test('when everything fits into a single mutation request', async () => { | ||
| const first = {createIfNotExists: {_id: 'foo', _type: 'something', bar: 'baz'}} | ||
| const second = {patch: {id: 'foo', set: {bar: 'baz'}}} | ||
| const gen = async function* () { | ||
| yield first | ||
| yield second | ||
| } | ||
|
|
||
| const firstSize = JSON.stringify(first).length | ||
| const secondSize = JSON.stringify(second).length | ||
|
|
||
| const it = batchMutations(gen(), firstSize + secondSize) | ||
|
|
||
| expect(await it.next()).toEqual({value: [first, second], done: false}) | ||
| expect(await it.next()).toEqual({value: undefined, done: true}) | ||
| }) | ||
|
|
||
| test('when max batch is not big enough to fit all values', async () => { | ||
| const first = {createIfNotExists: {_id: 'foo', _type: 'something', bar: 'baz'}} | ||
| const second = {patch: {id: 'foo', set: {bar: 'baz'}}} | ||
| const gen = async function* () { | ||
| yield first | ||
| yield second | ||
| } | ||
|
|
||
| const firstSize = JSON.stringify(first).length | ||
|
|
||
| const it = batchMutations(gen(), firstSize) | ||
|
|
||
| expect(await it.next()).toEqual({value: [first], done: false}) | ||
| expect(await it.next()).toEqual({value: [second], done: false}) | ||
| expect(await it.next()).toEqual({value: undefined, done: true}) | ||
| }) | ||
| test('when each mutation is smaller then max batch size', async () => { | ||
| const first = {createIfNotExists: {_id: 'foo', _type: 'something', bar: 'baz'}} | ||
| const second = {patch: {id: 'foo', set: {bar: 'baz'}}} | ||
| const gen = async function* () { | ||
| yield first | ||
| yield second | ||
| } | ||
|
|
||
| const it = batchMutations(gen(), 1) | ||
|
|
||
| expect(await it.next()).toEqual({value: [first], done: false}) | ||
| expect(await it.next()).toEqual({value: [second], done: false}) | ||
| expect(await it.next()).toEqual({value: undefined, done: true}) | ||
| }) | ||
|
|
||
| test('when some mutations can be chunked, others not', async () => { | ||
| const first = {createIfNotExists: {_id: 'foo', _type: 'something', bar: 'baz'}} | ||
| const second = {patch: {id: 'foo', set: {bar: 'baz'}}} | ||
|
|
||
| // Note: this is an array of mutations and should not be split up as it may be intentional to keep it in a transaction | ||
| // todo: is it ok to include other mutations in the same batch as a transaction? | ||
| const third = [ | ||
| { | ||
| createOrReplace: { | ||
| _id: 'foo', | ||
| _type: 'something', | ||
| bar: 'baz', | ||
| baz: 'qux', | ||
| }, | ||
| }, | ||
| {patch: {id: 'foo', set: {bar: 'baz'}}}, | ||
| ] | ||
|
|
||
| const gen = async function* () { | ||
| yield first | ||
| yield second | ||
| yield third | ||
| } | ||
| const it = batchMutations(gen(), byteLength(first) + byteLength(second)) | ||
|
|
||
| expect(await it.next()).toEqual({value: [first, second], done: false}) | ||
| expect(await it.next()).toEqual({value: third, done: false}) | ||
| }) | ||
| }) | ||
|
|
||
| //todo: verify if this is the default behavior we want | ||
| test('transactions are always in the same batch, but might include other mutations if they fit', async () => { | ||
| const first = {createIfNotExists: {_id: 'foo', _type: 'something', bar: 'baz'}} | ||
| const second = {patch: {id: 'foo', set: {bar: 'baz'}}} | ||
|
|
||
| // Note: this is an array of mutations and should not be split up as it may be intentional to keep it in a transaction | ||
| const transaction = [ | ||
| { | ||
| createOrReplace: { | ||
| _id: 'foo', | ||
| _type: 'something', | ||
| bar: 'baz', | ||
| baz: 'qux', | ||
| }, | ||
| }, | ||
| {patch: {id: 'foo', set: {bar: 'baz'}}}, | ||
| ] | ||
|
|
||
| const fourth = {patch: {id: 'another', set: {this: 'that'}}} | ||
|
|
||
| const gen = async function* () { | ||
| yield first | ||
| yield second | ||
| yield transaction | ||
| yield fourth | ||
| } | ||
| const it = batchMutations( | ||
| gen(), | ||
| [first, second, transaction, fourth].reduce((l, m) => l + byteLength(m), 0), | ||
| ) | ||
|
|
||
| expect(await it.next()).toEqual({value: [first, second, ...transaction, fourth], done: false}) | ||
| expect(await it.next()).toEqual({value: undefined, done: true}) | ||
| }) | ||
|
|
||
| test('transactions are always batched as-is if preserveTransactions: true', async () => { | ||
| const first = {createIfNotExists: {_id: 'foo', _type: 'something', bar: 'baz'}} | ||
| const second = {patch: {id: 'foo', set: {bar: 'baz'}}} | ||
|
|
||
| // Note: this is an array of mutations and should not be split up as it may be intentional to keep it in a transaction | ||
| const transaction = [ | ||
| { | ||
| createOrReplace: { | ||
| _id: 'foo', | ||
| _type: 'something', | ||
| bar: 'baz', | ||
| baz: 'qux', | ||
| }, | ||
| }, | ||
| {patch: {id: 'foo', set: {bar: 'baz'}}}, | ||
| ] | ||
|
|
||
| const fourth = {patch: {id: 'another', set: {this: 'that'}}} | ||
|
|
||
| const gen = async function* () { | ||
| yield first | ||
| yield second | ||
| yield transaction | ||
| yield fourth | ||
| } | ||
| const it = batchMutations( | ||
| gen(), | ||
| [first, second, transaction, fourth].reduce((l, m) => l + byteLength(m), 0), | ||
| {preserveTransactions: true}, | ||
| ) | ||
|
|
||
| expect(await it.next()).toEqual({value: [first, second], done: false}) | ||
| expect(await it.next()).toEqual({value: transaction, done: false}) | ||
| expect(await it.next()).toEqual({value: [fourth], done: false}) | ||
| expect(await it.next()).toEqual({value: undefined, done: true}) | ||
| }) |
55 changes: 55 additions & 0 deletions
55
packages/@sanity/migrate/src/runner/utils/batchMutations.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| import {Mutation as SanityMutation} from '@sanity/client' | ||
| import arrify from 'arrify' | ||
|
|
||
| // We're working on "raw" mutations, e.g what will be put into the mutations array in the request body | ||
| const PADDING_SIZE = '{"mutations":[]}'.length | ||
|
|
||
| /** | ||
| * | ||
| * @param mutations - Async iterable of either single values or arrays of values | ||
| * @param maxBatchSize - Max batch size in bytes | ||
| * Todo: add support for transaction ids too | ||
| */ | ||
| export async function* batchMutations( | ||
| mutations: AsyncIterableIterator<SanityMutation | SanityMutation[]>, | ||
| maxBatchSize: number, | ||
| options?: {preserveTransactions: boolean}, | ||
| ): AsyncIterableIterator<SanityMutation | SanityMutation[]> { | ||
| let currentBatch: SanityMutation[] = [] | ||
| let currentBatchSize = 0 | ||
|
|
||
| for await (const mutation of mutations) { | ||
| if (options?.preserveTransactions && Array.isArray(mutation)) { | ||
| yield currentBatch | ||
| yield mutation | ||
| currentBatch = [] | ||
| currentBatchSize = 0 | ||
| continue | ||
| } | ||
|
|
||
| // the mutation itself may exceed the payload size, need to handle that | ||
| const mutationSize = JSON.stringify(mutation).length | ||
|
|
||
| if (mutationSize >= maxBatchSize + PADDING_SIZE) { | ||
| // the mutation size itself is bigger then max batch size, yield it as a single batch and hope for the best (the server has a bigger limit) | ||
| if (currentBatch.length) { | ||
| yield currentBatch | ||
| } | ||
| yield [...arrify(mutation)] | ||
| currentBatch = [] | ||
| currentBatchSize = 0 | ||
| continue | ||
| } | ||
| currentBatchSize += mutationSize | ||
| if (currentBatchSize >= maxBatchSize + PADDING_SIZE) { | ||
| yield currentBatch | ||
| currentBatch = [] | ||
| currentBatchSize = 0 | ||
| } | ||
| currentBatch.push(...arrify(mutation)) | ||
| } | ||
|
|
||
| if (currentBatch.length > 0) { | ||
| yield currentBatch | ||
| } | ||
| } |
14 changes: 14 additions & 0 deletions
14
packages/@sanity/migrate/src/runner/utils/toSanityMutations.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| import {SanityEncoder, Mutation} from '@bjoerge/mutiny' | ||
| import {Mutation as SanityMutation} from '@sanity/client' | ||
|
|
||
| export async function* toSanityMutations( | ||
| it: AsyncIterableIterator<Mutation | Mutation[]>, | ||
| ): AsyncIterableIterator<SanityMutation | SanityMutation[]> { | ||
| for await (const mutation of it) { | ||
| if (Array.isArray(mutation)) { | ||
| yield SanityEncoder.encode(mutation) | ||
| continue | ||
| } | ||
| yield SanityEncoder.encode([mutation])[0] | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters