task: metadata: install: pip install -e .[dev]
patch: diff --git a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
index 9326992..21c1bab 100644
--- a/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
+++ b/google/cloud/pubsub_v1/subscriber/_protocol/streaming_pull_manager.py
@@ -988,7 +988,9 @@ class StreamingPullManager(object):
         # Return the initial request.
         return request
 
-    def _send_lease_modacks(self, ack_ids: Iterable[str], ack_deadline: float):
+    def _send_lease_modacks(
+        self, ack_ids: Iterable[str], ack_deadline: float
+    ) -> List[str]:
         exactly_once_enabled = False
         with self._exactly_once_enabled_lock:
             exactly_once_enabled = self._exactly_once_enabled
@@ -1002,15 +1004,19 @@ class StreamingPullManager(object):
             assert self._dispatcher is not None
             self._dispatcher.modify_ack_deadline(items)
 
+            expired_ack_ids = []
             for req in items:
                 try:
                     assert req.future is not None
                     req.future.result()
-                except AcknowledgeError:
+                except AcknowledgeError as ack_error:
                     _LOGGER.warning(
                         "AcknowledgeError when lease-modacking a message.",
                         exc_info=True,
                     )
+                    if ack_error.error_code == AcknowledgeStatus.INVALID_ACK_ID:
+                        expired_ack_ids.append(req.ack_id)
+            return expired_ack_ids
         else:
             items = [
                 requests.ModAckRequest(ack_id, self.ack_deadline, None)
@@ -1018,6 +1024,7 @@ class StreamingPullManager(object):
             ]
             assert self._dispatcher is not None
             self._dispatcher.modify_ack_deadline(items)
+            return []
 
     def _exactly_once_delivery_enabled(self) -> bool:
         """Whether exactly-once delivery is enabled for the subscription."""
@@ -1071,28 +1078,32 @@ class StreamingPullManager(object):
         # modack the messages we received, as this tells the server that we've
         # received them.
         ack_id_gen = (message.ack_id for message in received_messages)
-        self._send_lease_modacks(ack_id_gen, self.ack_deadline)
+        expired_ack_ids = set(self._send_lease_modacks(ack_id_gen, self.ack_deadline))
 
         with self._pause_resume_lock:
             assert self._scheduler is not None
             assert self._leaser is not None
 
             for received_message in received_messages:
-                message = google.cloud.pubsub_v1.subscriber.message.Message(
-                    received_message.message,
-                    received_message.ack_id,
-                    received_message.delivery_attempt,
-                    self._scheduler.queue,
-                    self._exactly_once_delivery_enabled,
-                )
-                self._messages_on_hold.put(message)
-                self._on_hold_bytes += message.size
-                req = requests.LeaseRequest(
-                    ack_id=message.ack_id,
-                    byte_size=message.size,
-                    ordering_key=message.ordering_key,
-                )
-                self._leaser.add([req])
+                if (
+                    not self._exactly_once_delivery_enabled()
+                    or received_message.ack_id not in expired_ack_ids
+                ):
+                    message = google.cloud.pubsub_v1.subscriber.message.Message(
+                        received_message.message,
+                        received_message.ack_id,
+                        received_message.delivery_attempt,
+                        self._scheduler.queue,
+                        self._exactly_once_delivery_enabled,
+                    )
+                    self._messages_on_hold.put(message)
+                    self._on_hold_bytes += message.size
+                    req = requests.LeaseRequest(
+                        ack_id=message.ack_id,
+                        byte_size=message.size,
+                        ordering_key=message.ordering_key,
+                    )
+                    self._leaser.add([req])
 
             self._maybe_release_messages()
 

problem_statement: Remove expired ack_ids from lease management.
Possible root cause of: https://github.com/googleapis/python-pubsub/issues/593

#### Steps to reproduce

  1. High-latency subscriber (e.g. 2 hours)
  2. Low message count flow control
  3. Solution only works for exactly once delivery subscriptions, since they return invalid_ack_id errors.

#### Code example

```

def f(x):
    set_time = 60
    timeout = time.time() + 60*float(set_time)  # X minutes from now
    print(timeout)
    while True:
        if time.time() > timeout:
            print("Time to ack!")
            print(time.time())
            break
        x * x

def create_load(percent_load):
    processes = psutil.cpu_count()
    to_use = int(processes * (percent_load/100))
    print ('utilizing %d cores\n' % to_use)
    pool = Pool(to_use)
    pool.map(f, range(to_use))

def receive_messages_with_flow_control() -> None:
    """Receives messages from a pull subscription with flow control."""
    from concurrent.futures import TimeoutError
    from google.cloud import pubsub_v1

    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(project_id, subscription_id)

    def callback(message: pubsub_v1.subscriber.message.Message) -> None:
        print(f"Received {message.data!r} at {time.time()}.")
        create_load(50)
        message.ack()

    # Limit the subscriber to only have ten outstanding messages at a time.
    flow_control = pubsub_v1.types.FlowControl(max_messages=1)

    streaming_pull_future = subscriber.subscribe(
        subscription_path, callback=callback, flow_control=flow_control
    )

    # Wrap subscriber in a 'with' block to automatically call close() when done.
    with subscriber:
        try:
            # When `timeout` is not set, result() will block indefinitely,
            # unless an exception is encountered first.
            streaming_pull_future.result()
        except TimeoutError:
            streaming_pull_future.cancel()  # Trigger the shutdown.
            streaming_pull_future.result()  # Block until the shutdown is complete.
```

### Error trace:
```
google.api_core.exceptions.InvalidArgument: 400 Some acknowledgement ids in the request were invalid. This could be because the acknowledgement ids have expired or the acknowledgement ids were malformed. [reason: "EXACTLY_ONCE_ACKID_FAILURE"
domain: "pubsub.googleapis.com"
metadata {
  value: "PERMANENT_FAILURE_INVALID_ACK_ID"
}
```

Even while flow control is activate and CPS client library does not receive messages in streaming_pull_manager._on_response, CPS still delivers messages to the bidi stream, which are buffered. These messages are not lease managed (because the library does not know about them) , so they expire before they are ever sent to a callback.

When one of these expired messages is passed to _on_response, and a receipt modack is sent via send_lease_modacks, if we receive an EOD related INVALID_ACK_ID failure, we should not pass these messages on to be processed, as they are already expired.

repo: googleapis/python-pubsub
test_cmd: pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
test_patch: diff --git a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
index 459ab6d..b4f76f2 100644
--- a/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
+++ b/tests/unit/pubsub_v1/subscriber/test_streaming_pull_manager.py
@@ -1125,6 +1125,7 @@ def test_heartbeat_stream_ack_deadline_seconds(caplog):
     "google.cloud.pubsub_v1.subscriber._protocol.heartbeater.Heartbeater", autospec=True
 )
 def test_open(heartbeater, dispatcher, leaser, background_consumer, resumable_bidi_rpc):
+
     manager = make_manager()
 
     with mock.patch.object(
@@ -1852,11 +1853,18 @@ def test__on_response_exactly_once_immediate_modacks_fail():
     def complete_futures_with_error(*args, **kwargs):
         modack_requests = args[0]
         for req in modack_requests:
-            req.future.set_exception(
-                subscriber_exceptions.AcknowledgeError(
-                    subscriber_exceptions.AcknowledgeStatus.SUCCESS, None
+            if req.ack_id == "fack":
+                req.future.set_exception(
+                    subscriber_exceptions.AcknowledgeError(
+                        subscriber_exceptions.AcknowledgeStatus.INVALID_ACK_ID, None
+                    )
+                )
+            else:
+                req.future.set_exception(
+                    subscriber_exceptions.AcknowledgeError(
+                        subscriber_exceptions.AcknowledgeStatus.SUCCESS, None
+                    )
                 )
-            )
 
     dispatcher.modify_ack_deadline.side_effect = complete_futures_with_error
 
@@ -1866,19 +1874,39 @@ def test__on_response_exactly_once_immediate_modacks_fail():
             gapic_types.ReceivedMessage(
                 ack_id="fack",
                 message=gapic_types.PubsubMessage(data=b"foo", message_id="1"),
-            )
+            ),
+            gapic_types.ReceivedMessage(
+                ack_id="good",
+                message=gapic_types.PubsubMessage(data=b"foo", message_id="2"),
+            ),
         ],
         subscription_properties=gapic_types.StreamingPullResponse.SubscriptionProperties(
             exactly_once_delivery_enabled=True
         ),
     )
 
-    # adjust message bookkeeping in leaser
-    fake_leaser_add(leaser, init_msg_count=0, assumed_msg_size=42)
+    # Actually run the method and prove that modack and schedule are called in
+    # the expected way.
+
+    fake_leaser_add(leaser, init_msg_count=0, assumed_msg_size=10)
 
-    # exactly_once should be enabled
     manager._on_response(response)
-    # exceptions are logged, but otherwise no effect
+
+    # The second messages should be scheduled, and not the first.
+
+    schedule_calls = scheduler.schedule.mock_calls
+    assert len(schedule_calls) == 1
+    call_args = schedule_calls[0][1]
+    assert call_args[0] == mock.sentinel.callback
+    assert isinstance(call_args[1], message.Message)
+    assert call_args[1].message_id == "2"
+
+    assert manager._messages_on_hold.size == 0
+    # No messages available
+    assert manager._messages_on_hold.get() is None
+
+    # do not add message
+    assert manager.load == 0.001
 
 
 def test__should_recover_true():