[boto3sqs] delete_message_batch returns inside its loop, so it handles the first entry only

Describe your environment

• opentelemetry-instrumentation-boto3sqs 0.65b0, and on main at b4ba6ca
• Python 3.12.3, boto3 and botocore 1.34.44
• No AWS account needed: the reproduction mocks the botocore endpoint

What happened?

delete_message_wrapper_batch puts its return inside the for loop, so the loop can only ever run
one iteration and every entry after the first is unreachable:

https://github.com/open-telemetry/opentelemetry-python-contrib/blob/b4ba6ca49/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py#L305-L311

Three consequences, measured on b4ba6ca. A consumer that polls the maximum of ten messages and
deletes them in one call sees all three:

1. Every entry after the first keeps an unended process span, and an unended span is never exported.
   That batch of ten exports 1 of the 10.
2. Each of those spans stays in received_messages_spans, which is a class attribute that
   _uninstrument does not clear. Eleven such polls leave 99 entries.
3. The context token that ContextableList.__getitem__ attached is never detached, so the next span
   started in that thread is parented to a span nothing exports. In the reproduction below, an
   unrelated start_as_current_span after the batch delete gets the retained rh-2 span as its
   parent. With the return dedented it has no parent.

Steps to Reproduce

pip install opentelemetry-instrumentation-boto3sqs opentelemetry-sdk 'boto3==1.34.44'
python repro.py

from unittest import mock

import boto3
from botocore.awsrequest import AWSResponse
from opentelemetry import trace
from opentelemetry.instrumentation.boto3sqs import Boto3SQSInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
Boto3SQSInstrumentor().instrument(tracer_provider=provider)

QUEUE_URL = "https://sqs.us-east-1.amazonaws.com/123456789012/MyQueue"
client = boto3.client("sqs", region_name="us-east-1", aws_access_key_id="x", aws_secret_access_key="x")


def fake(response):
    return lambda *a, **k: (AWSResponse("http://127.0.0.1", 200, {}, "{}"), response)


messages = [
    {"MessageId": "1", "ReceiptHandle": "rh-1", "Body": "a", "MessageAttributes": {}},
    {"MessageId": "2", "ReceiptHandle": "rh-2", "Body": "b", "MessageAttributes": {}},
]
with mock.patch("botocore.endpoint.Endpoint.make_request", new=fake({"Messages": messages})):
    response = client.receive_message(QueueUrl=QUEUE_URL)
for message in response["Messages"]:
    pass  # processing
with mock.patch("botocore.endpoint.Endpoint.make_request", new=fake({"Successful": [], "Failed": []})):
    client.delete_message_batch(
        QueueUrl=QUEUE_URL,
        Entries=[{"Id": "1", "ReceiptHandle": "rh-1"}, {"Id": "2", "ReceiptHandle": "rh-2"}],
    )

with trace.get_tracer("app").start_as_current_span("unrelated-work"):
    pass

finished = [s.name for s in exporter.get_finished_spans() if s.name.endswith(" process")]
unrelated = [s for s in exporter.get_finished_spans() if s.name == "unrelated-work"][0]
retained = Boto3SQSInstrumentor.received_messages_spans.get("rh-2")
print("process spans ended by delete_message_batch:", len(finished))
print("still in received_messages_spans:", sorted(Boto3SQSInstrumentor.received_messages_spans))
print(
    "unrelated-work is parented to the retained rh-2 span:",
    retained is not None
    and unrelated.parent is not None
    and unrelated.parent.span_id == retained.get_span_context().span_id,
)

Expected Result

process spans ended by delete_message_batch: 2
still in received_messages_spans: []
unrelated-work is parented to the retained rh-2 span: False

Actual Result

process spans ended by delete_message_batch: 1
still in received_messages_spans: ['rh-2']
unrelated-work is parented to the retained rh-2 span: True

Additional context

The sibling delete_message wrapper at :295-299 has the same body and calls wrapped once, after
the loop-free if, which is where the batch version's return belongs. Dedenting it by one level
gives the expected output, and the suite passes: 16 passed, the same count as before the change. No test exercises
delete_message_batch beyond asserting that the wrapper exists
(tests/test_boto3sqs_instrumentation.py:54), so that count is not evidence and a regression test
belongs with the fix.

The same six lines swallow two calls that never reach botocore, because entries is read once and
never guarded. Entries=[] skips the loop body, so the request is not sent and the caller gets None
where botocore returns a dict. A positional call raises TypeError: 'NoneType' object is not iterable
from inside the wrapper, where botocore alone says delete_message_batch() only accepts keyword arguments. The dedent fixes the first. The second wants a guard, which changes an exception type, so
it belongs in a separate commit.

One decision to settle before that: should the fix end a span for every entry the caller requested, or
only for the ids the response returns under Successful? Today's code ends spans before wrapped
runs and never reads the response, so ending them all keeps the current semantics. Reading
Successful instead would be a behaviour change, and it needs saying out loud.

Not #1704, which I commented on last week. That issue reports entries that leak when a message is
never deleted, and it lists this path as one that removes them. This one fires when the message is
deleted, in one successful call. #2509 lists five design problems and this is not among them. #5056 is
an open pull request on this file and leaves these lines byte-identical at its head d3d52cd, so the
two do not collide: I can send the fix now and rebase whichever lands second. The nearest precedent is
#4744, the same file and the sibling send_message_batch wrapper, fixed by #4746 in fourteen days.

I ran the wrapper against a mocked endpoint, not a real queue, and I did not check whether SQS accepts
an empty Entries list. Uninstrumented, botocore sends the request either way.

Would you like to implement a fix?

Yes

Start in instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/init.py around delete_message_wrapper_batch and compare it with the sibling delete_message wrapper. Add regression coverage in tests/test_boto3sqs_instrumentation.py, then run that test file; done means a batch with two entries ends both process spans, clears received_messages_spans, and does not retain the deleted span as parent.