Cloudflare Developers•4w ago

Serialized RPC arguments or return values are limited to 1MiB

When trying to move to a Service Binding oriented architecture between workers, I bumped into the following error: Serialized RPC arguments or return values are limited to 1MiB This is happening when trying to transmit a large dataset through a service binding to the worker that handles DB operations. The array of data (trimmed as much as possible to an array of arrays) is just over 1 MB in size. It feels strange that I may need to break this data up into multiple RPC calls to get past this limitation when doing it over HTTP worked just fine. 1. Is it possible to up this limit for a worker? 2. Do I have other options? Thanks!

2 Replies

DaniFoldi•4w ago

Hi, I think you should be able to pass around larger amounts of data with streams - although I haven't specifically tested the size limit, they should work with any size, even larger than the worker memory size (unbuffered of course, which doesn't count here). So if instead of using an array as the argument, you pass in a readablestream, you should be able to read from the service binding what you pipe in from your caller and vice versa.

Murder Chicken•4w ago

@Purple Blob thanks for the input. Are you aware of any code examples of how to do with with RPC service bindings? Also, the documentation notes that the Stream API is only available within a fetch event. Do I have this or other options within a scheduled event? For anyone who stumbles into this thread... I managed to figure this out (thanks Cursor IDE). Worker making the RPC call with a large payload...

const data = [[data], [data], [data], ...];
const encoder = new TextEncoder();
const stream = new ReadableStream({
  start(controller) {
    data.forEach(chunk => {
      const bytes = encoder.encode(JSON.stringify(chunk) + '\n');
      controller.enqueue(bytes);
    });
    controller.close();
  }
});

// Send to other worker RPC method through Service Binding...
await env.SERVICE_BINDING.rpcMethod(stream);

const data = [[data], [data], [data], ...];
const encoder = new TextEncoder();
const stream = new ReadableStream({
  start(controller) {
    data.forEach(chunk => {
      const bytes = encoder.encode(JSON.stringify(chunk) + '\n');
      controller.enqueue(bytes);
    });
    controller.close();
  }
});

// Send to other worker RPC method through Service Binding...
await env.SERVICE_BINDING.rpcMethod(stream);

Consumer worker with RPC method...

async rpcMethod(stream) {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  let data = [];
  
  try {
    while (true) {
      let { done, value } = await reader.read().catch(error => {
        console.warn('Stream read error:', error);
        return { done: true, value: undefined };
      });
  
      if (done) {
        if (buffer) {
          try {
            data.push(JSON.parse(buffer));
          } catch (error) {
            console.warn('Error parsing remaining buffer:', error);
          }
        }
        break;
      }
  
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop();
  
      for (const line of lines) {
        if (line) {
          try {
            data.push(JSON.parse(line));
          } catch (error) {
            console.warn('Error parsing JSON:', error, 'Line:', line);
          }
        }
      }
    }
  } catch (error) {
    console.error('Stream processing error:', error);
  } finally {
    reader.releaseLock();
  }
  
  // Do whatever with the data
}

async rpcMethod(stream) {
  const reader = stream.getReader();
  const decoder = new TextDecoder();
  let buffer = '';
  let data = [];
  
  try {
    while (true) {
      let { done, value } = await reader.read().catch(error => {
        console.warn('Stream read error:', error);
        return { done: true, value: undefined };
      });
  
      if (done) {
        if (buffer) {
          try {
            data.push(JSON.parse(buffer));
          } catch (error) {
            console.warn('Error parsing remaining buffer:', error);
          }
        }
        break;
      }
  
      buffer += decoder.decode(value, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop();
  
      for (const line of lines) {
        if (line) {
          try {
            data.push(JSON.parse(line));
          } catch (error) {
            console.warn('Error parsing JSON:', error, 'Line:', line);
          }
        }
      }
    }
  } catch (error) {
    console.error('Stream processing error:', error);
  } finally {
    reader.releaseLock();
  }
  
  // Do whatever with the data
}

Gaming

Programming

Serialized RPC arguments or return values are limited to 1MiB