in_s3: stop retrying SQS messages that cannot be decoded

[Watson1978]

The problem

Posting hello to the queue is enough to stall an in_s3 source indefinitely.

is_valid_queue only checks that the top-level Records (or detail) key exists, so a body without the nested s3.object.key gets through it and raises in get_raw_key. The rescue in run turns every per-message error into :skip_delete, which tells QueuePoller to leave the message in the queue. SQS redelivers it after the visibility timeout, it fails identically, and this repeats until the retention period expires — up to 14 days of CPU, log volume and SQS requests, with a backtrace logged on every pass and nothing ever ingested. skip_delete is false by default, so no configuration is needed to reach this, and anything that can put a message on the queue can trigger it.

The change

• Validate a message down to its object key before touching S3, and skip the ones that cannot name an object — a body that is not JSON, is not an object, is not an S3 notification, has no usable key, or has a key that decodes to invalid UTF-8. Those can never succeed on redelivery, so they are skipped and deleted instead of retried forever
• Leave every failure after the key is known — a failed get, a corrupt archive, a parse error — on :skip_delete. The object exists there, a misconfigured s3_bucket would otherwise drain the whole backlog, and a redrive policy with a dead letter queue is the right tool for it
• Keep the decoding inside the per-message rescue, so an unforeseen error still falls back to :skip_delete rather than escaping QueuePoller#poll and halting ingestion for retry_error_interval seconds
• Log one warn per skipped message so a mistake stays visible: an SNS envelope, or an event_bridge_mode that does not match the queue, used to be dropped in silence and looked exactly like an idle queue
• Route everything logged through one helper that truncates to 256 bytes and scrubs, because an object key can be megabytes long and can hold invalid UTF-8 that JSON.generate rejects when the log event record is built

Notes

process now takes the object key rather than the parsed body. Tests go from 108 to 143, and they redeliver the same message to prove a skipped one is handled exactly once while a failed fetch is still retried. Two pre-existing limitations are unchanged and left for separate changes: only Records[0] is examined, and add_object_metadata reports the URL-escaped key.