Merge pull request activepieces#5255 from activepieces/fix/store

fix: enforce store limit

docs/flows/known-limits.mdx

@@ -0,0 +1,35 @@
+---
+title: "Technical limits"
+icon: 'ban'
+description: "technical limits for Activepieces execution"
+---
+
+### Overview
+
+<Warning>
+This Limits applies for the **Activepieces Cloud**, and can be configured via environment variables for self-hosted instances.
+</Warning>
+
+### Flow Limits
+
+- **Execution Time**: Each flow has a maximum execution time of **600 seconds (10 minutes)**. Flows exceeding this limit will be marked as a timeout.
+- **Memory Usage**: During execution, a flow should not use more than **128 MB of RAM**.
+
+<Tip>
+**Friendly Tip #1:** Flow run in a paused state, such as Wait for Approval or Delay, do not count toward the 600 seconds.
+</Tip>
+
+<Tip>
+**Friendly Tip #2:** The execution time limit can be worked around by splitting the flows into multiple ones, such as by having one flow call another flow using a webhook, or by having each flow process a small batch of items.
+</Tip>
+
+### Data Storage Limits
+
+Some pieces utilize the built-in Activepieces key store, such as the Store Piece and Queue Piece.
+
+The storage limits are as follows:
+
+- **Maximum Key Length**: 128 characters
+- **Maximum Value Size**: 512 KB
+
+

docs/mint.json

@@ -174,7 +174,8 @@
         "flows/passing-data",
         "flows/publishing-flows",
         "flows/debugging-runs",
-        "flows/versioning"
+        "flows/versioning",
+        "flows/known-limits"
       ]
     },
     {

packages/engine/src/lib/services/storage.service.ts

@@ -2,6 +2,7 @@ import { URL } from 'node:url'
 import { Store, StoreScope } from '@activepieces/pieces-framework'
 import { DeleteStoreEntryRequest, FlowId, PutStoreEntryRequest, STORE_VALUE_MAX_SIZE, StoreEntry } from '@activepieces/shared'
 import { StatusCodes } from 'http-status-codes'
+import sizeof from 'object-sizeof'
 import { ExecutionError, FetchError, StorageError, StorageLimitError } from '../helper/execution-errors'
 
 export const createStorageService = ({ engineToken, apiUrl }: CreateStorageServiceParams): StorageService => {
@@ -37,6 +38,10 @@ export const createStorageService = ({ engineToken, apiUrl }: CreateStorageServi
             const url = buildUrl(apiUrl)
 
             try {
+                const sizeOfValue = sizeof(request.value)
+                if (sizeOfValue > STORE_VALUE_MAX_SIZE) {
+                    throw new StorageLimitError(request.key, STORE_VALUE_MAX_SIZE)
+                }
                 const response = await fetch(url, {
                     method: 'POST',
                     headers: {

packages/pieces/community/queue/package.json

@@ -1,4 +1,4 @@
 {
   "name": "@activepieces/piece-queue",
-  "version": "0.0.5"
+  "version": "0.0.6"
 }

packages/server/api/src/app/store-entry/store-entry.controller.ts

@@ -3,83 +3,73 @@ import {
     GetStoreEntryRequest,
     PrincipalType,
     PutStoreEntryRequest,
+    STORE_VALUE_MAX_SIZE,
 } from '@activepieces/shared'
 import { FastifyPluginAsyncTypebox } from '@fastify/type-provider-typebox'
-import { FastifyRequest } from 'fastify'
 import { StatusCodes } from 'http-status-codes'
+import sizeof from 'object-sizeof'
 import { storeEntryService } from './store-entry.service'
 
 export const storeEntryController: FastifyPluginAsyncTypebox = async (
     fastify,
 ) => {
-    fastify.post(
-        '/',
-        {
-            schema: {
-                body: PutStoreEntryRequest,
-            },
-        },
-        async (
-            request: FastifyRequest<{
-                Body: PutStoreEntryRequest
-            }>,
-            _reply,
-        ) => {
-            return storeEntryService.upsert({
-                projectId: request.principal.projectId,
-                request: request.body,
-            })
-        },
+    fastify.post( '/', CreateRequest, async (request, reply) => {
+        const sizeOfValue = sizeof(request.body.value)
+        if (sizeOfValue > STORE_VALUE_MAX_SIZE) {
+            await reply.status(StatusCodes.REQUEST_TOO_LONG).send({})
+            return
+        }
+        const response = await storeEntryService.upsert({
+            projectId: request.principal.projectId,
+            request: request.body,
+        })
+        await reply.status(StatusCodes.OK).send(response)
+    },
     )
 
-    fastify.get(
-        '/',
-        {
-            schema: {
-                querystring: GetStoreEntryRequest,
-            },
-        },
-        async (
-            request: FastifyRequest<{
-                Querystring: GetStoreEntryRequest
-            }>,
-            reply,
-        ) => {
-            const value = await storeEntryService.getOne({
-                projectId: request.principal.projectId,
-                key: request.query.key,
-            })
+    fastify.get('/', GetRequest, async (request, reply) => {
+        const value = await storeEntryService.getOne({
+            projectId: request.principal.projectId,
+            key: request.query.key,
+        })
 
-            if (!value) {
-                return reply.code(StatusCodes.NOT_FOUND).send('Value not found!')
-            }
+        if (!value) {
+            return reply.code(StatusCodes.NOT_FOUND).send('Value not found!')
+        }
 
-            return value
-        },
+        return value
+    },
     )
 
-    fastify.delete(
-        '/',
-        {
-            schema: {
-                querystring: DeleteStoreEntryRequest,
-            },
-        },
-        async (
-            request: FastifyRequest<{
-                Querystring: DeleteStoreEntryRequest
-            }>,
-            reply,
-        ) => {
-            if (request.principal.type !== PrincipalType.ENGINE) {
-                return reply.status(StatusCodes.FORBIDDEN)
-            }
-            else {
-                return storeEntryService.delete({
-                    projectId: request.principal.projectId,
-                    key: request.query.key,
-                })
-            }
-        },
+    fastify.delete( '/', DeleteStoreRequest, async (request, reply) => {
+        if (request.principal.type !== PrincipalType.ENGINE) {
+            return reply.status(StatusCodes.FORBIDDEN)
+        }
+        else {
+            return storeEntryService.delete({
+                projectId: request.principal.projectId,
+                key: request.query.key,
+            })
+        }
+    },
     )
 }
+
+const CreateRequest =  {
+    schema: {
+        body: PutStoreEntryRequest,
+    },
+}
+
+const GetRequest = {
+    schema: {
+        querystring: GetStoreEntryRequest,
+    },
+}
+
+
+const DeleteStoreRequest = {
+    schema: {
+        querystring: DeleteStoreEntryRequest,
+    },
+}

packages/shared/src/lib/store-entry/store-entry.ts

@@ -5,8 +5,7 @@ import { ProjectId } from '../project/project'
 export type StoreEntryId = ApId
 
 export const STORE_KEY_MAX_LENGTH = 128
-
-export const STORE_VALUE_MAX_SIZE = 4 * 1024 * 1024
+export const STORE_VALUE_MAX_SIZE = 512 * 1024
 
 export type StoreEntry = {
     key: string