diff --git a/docker-compose.yml b/docker-compose.yml index 101f5df..d0d459e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -24,6 +24,8 @@ services: - AWS_DEFAULT_REGION=us-east-1 - AWS_ACCESS_KEY_ID=dummy - AWS_SECRET_ACCESS_KEY=dummy + volumes: + - "./test:/usr/test" depends_on: localstack-persist: condition: service_healthy diff --git a/src/localstack_persist/state.py b/src/localstack_persist/state.py index d5c2772..a4ce735 100644 --- a/src/localstack_persist/state.py +++ b/src/localstack_persist/state.py @@ -10,7 +10,7 @@ from localstack.services.plugins import SERVICE_PLUGINS from localstack.aws.api import RequestContext from collections import defaultdict -from threading import Thread, Condition +from threading import Thread, Condition, Timer from readerwriterlock.rwlock import RWLockWrite, Lockable from .visitors import LoadStateVisitor, SaveStateVisitor from .config import BASE_DIR, is_persistence_enabled, PERSIST_FREQUENCY @@ -71,10 +71,16 @@ def on_request(self, chain, context: RequestContext, response): if service_name not in self.loaded_services: self._load_service_state(service_name) - # Prevent persistence from running for this service while handling this request + # Prevent persistence from running for this service while handling this request... rlock = self.rwlocks[service_name].gen_rlock() setattr(context, "localstack-persist_rlock", rlock) rlock.acquire() + # ...unless the request takes over 1 second, in which case we force release the lock to + # prevent long-running requests from blocking persistence which would in turn block other + # requests + timer = Timer(1, try_release, [rlock]) + setattr(context, "localstack-persist_rlock_timer", timer) + timer.start() def on_response(self, chain, context: RequestContext, response): if not context.service or not context.request or not context.operation: @@ -94,8 +100,15 @@ def on_response(self, chain, context: RequestContext, response): self.add_affected_service(service_name) def on_finalize(self, chain, context: RequestContext, response): - if rlock := getattr(context, "localstack-persist_rlock", None): - cast(Lockable, rlock).release() + if rlock := cast( + Lockable | None, getattr(context, "localstack-persist_rlock", None) + ): + try_release(rlock) + + if timer := cast( + Timer | None, getattr(context, "localstack-persist_rlock_timer", None) + ): + timer.cancel() def load_all_services_state(self): LOG.info("Loading persisted state of all services...") @@ -182,3 +195,11 @@ def _save_service_state(self, service_name: str): STATE_TRACKER = StateTracker() + + +def try_release(lock: Lockable): + if lock and lock.locked(): + try: + lock.release() + except: + pass diff --git a/test/Dockerfile b/test/Dockerfile index 2ac23fe..f8db352 100644 --- a/test/Dockerfile +++ b/test/Dockerfile @@ -1,7 +1,7 @@ FROM python -RUN pip3 install boto3==1.28.80 botocore==1.31.80 +WORKDIR /usr/test -COPY . . +RUN pip3 install boto3==1.28.80 botocore==1.31.80 ENTRYPOINT [ "python", "./main.py" ] \ No newline at end of file diff --git a/test/main.py b/test/main.py index 0226b0d..ace86d6 100644 --- a/test/main.py +++ b/test/main.py @@ -107,8 +107,8 @@ def wait_until_es_ready(domain_name: str): assert_equal(queue.attributes["ApproximateNumberOfMessages"], "0") table = dynamodb.Table("test-table") - item = table.get_item(Key={"id": 123})["Item"] - assert_equal(item["foo"], "bar") + item = table.get_item(Key={"id": 123}).get("Item", {}) + assert_equal(item.get("foo"), "bar") bucket = s3.Bucket("test-bucket") obj = bucket.Object("test-object")