take3. fix for slow LocalStack/py 3.14 (#60)

Author: hampsterx

kinesis/base.py

@@ -68,6 +68,7 @@ def __init__(
         self._reconnect_timeout = time.monotonic()
         self.create_stream = create_stream
         self.create_stream_shards = create_stream_shards
+        self._just_created = False
 
     async def __aenter__(self) -> "Base":
 
@@ -158,16 +159,16 @@ async def start(self):
 
         await self.get_client()
 
-        just_created = False
         if self.create_stream:
             await self._create_stream()
             self.create_stream = False
-            just_created = True
+            self._just_created = True
 
         if self.skip_describe_stream:
             log.debug("Skipping Describe stream '{}'. Assuming it exists and is active.".format(self.stream_name))
             self.shards = []
             self.stream_status = self.ACTIVE
+            self._just_created = False
             return
 
         log.debug("Checking stream '{}' is active".format(self.stream_name))
@@ -178,6 +179,7 @@ async def start(self):
             try:
                 self.shards = await self.list_shards()
                 self.stream_status = self.ACTIVE
+                self._just_created = False
             except Exception as e:
                 # Fall back to DescribeStream if ListShards fails
                 log.warning(f"ListShards failed ({e}), falling back to DescribeStream")
@@ -191,7 +193,7 @@ async def start(self):
                         try:
                             stream_info = await self.get_stream_description()
                         except exceptions.StreamDoesNotExist:
-                            if just_created:
+                            if self._just_created:
                                 # Stream was just created but may not be visible yet (eventual consistency)
                                 log.debug(
                                     "Stream '{}' not yet visible after creation, retrying...".format(self.stream_name)
@@ -203,6 +205,7 @@ async def start(self):
 
                         if stream_status == self.ACTIVE:
                             self.stream_status = stream_status
+                            self._just_created = False
                             break
 
                         if stream_status in ["CREATING", "UPDATING"]:

tests/test_producer.py

@@ -594,3 +594,44 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
             # The key test is that the code runs without errors and includes partition key in calculation
             assert throttle_sizes[0] >= 1, "Should include partition key in size calculation"
             assert throttle_sizes[1] >= 1, "Should calculate size for item without partition key"
+
+    @pytest.mark.asyncio
+    async def test_just_created_survives_reconnection(self, random_stream_name, endpoint_url):
+        """Test that _just_created state persists across reconnection attempts (Issue #56).
+
+        When create_stream=True and the backend is slow to make the stream visible,
+        start() sets self.create_stream = False after _create_stream(). If the 60s
+        describe timeout expires and get_conn() triggers a reconnection, the second
+        start() call must still know the stream was just created so it retries
+        StreamDoesNotExist instead of propagating immediately.
+        """
+        producer = Producer(
+            stream_name=random_stream_name,
+            endpoint_url=endpoint_url,
+            region_name="ap-southeast-2",
+            create_stream=True,
+            create_stream_shards=1,
+        )
+
+        assert producer._just_created is False
+        assert producer.create_stream is True
+
+        # Simulate what start() does: create stream, flip flags
+        await producer.get_client()
+        await producer._create_stream()
+        producer.create_stream = False
+        producer._just_created = True
+
+        # After first start() sets create_stream=False, a reconnection calls start() again.
+        # _just_created must still be True so the retry logic handles StreamDoesNotExist.
+        assert producer.create_stream is False
+        assert producer._just_created is True
+
+        # Now simulate a successful second start() — stream becomes ACTIVE
+        await producer.start()
+
+        # After stream confirmed ACTIVE, _just_created should be reset
+        assert producer._just_created is False
+        assert producer.stream_status == producer.ACTIVE
+
+        await producer.close()