python/etl: classify direct-put responses by Ais-Direct-Put-Complete marker

python/CHANGELOG.md

@@ -47,6 +47,18 @@ We structure this changelog in accordance with [Keep a Changelog](https://keepac
 
 ### Fixed
 
+- ETL webservers now classify direct-put responses by the new
+  `Ais-Direct-Put-Complete` marker header, which the AIS target returns
+  (alongside `204` and `Ais-Direct-Put-Length`) after storing an object via
+  the direct-PUT endpoint. A marked ack is passed back through the pipeline
+  as-is: the 204, the marker, and the length (even when the length is 0 for
+  an empty stored object). Markerless responses use a `Content-Length`-keyed
+  fallback — `0` means delivered, while absent (chunked) or non-zero means
+  transformed content, forwarded as-is, so a chunked 200 with an empty body
+  (a valid empty transform result) is no longer misreported as delivered.
+  The fallback exists for targets that predate the marker and will be phased
+  out with them. `ETLServer.handle_direct_put_response` now returns a
+  4-tuple `(status, body, direct_put_length, direct_put_complete)`.
 - ETL webservers now forward
   `etl_args` to the next stage on direct-put pipeline hops. Previously only the
   first pipeline stage received `etl_args`; stages 2..N saw an empty value.

python/aistore/sdk/const.py

@@ -61,6 +61,11 @@
 HEADER_OBJECT_BLOB_WORKERS = HEADER_PREFIX + "Blob-Workers"
 HEADER_OBJECT_APPEND_HANDLE = HEADER_PREFIX + "Append-Handle"
 HEADER_DIRECT_PUT_LENGTH = HEADER_PREFIX + "Direct-Put-Length"
+# Set by the AIS target (alongside a 204 and HEADER_DIRECT_PUT_LENGTH) after a
+# successful direct PUT to signal that the destination already stored the
+# object. Presence-based (value ignored); intermediate ETL webservers propagate
+# it back through the pipeline.
+HEADER_DIRECT_PUT_COMPLETE = HEADER_PREFIX + "Direct-Put-Complete"
 # ETL → AIS retry contract: emitted by the ETL webserver alongside a 503
 # response to signal that AIS should retry the whole PUT (the ETL bailed on
 # a transient direct-put failure and the request body was one-shot).

python/aistore/sdk/etl/webserver/base_etl_server.py

@@ -16,6 +16,8 @@
     STATUS_OK,
     STATUS_BAD_GATEWAY,
     HEADER_AUTHORIZATION,
+    HEADER_CONTENT_LENGTH,
+    HEADER_DIRECT_PUT_COMPLETE,
     HEADER_DIRECT_PUT_LENGTH,
     AIS_AUTHN_TOKEN,
     AIS_DIRECT_PUT_RETRIES,
@@ -64,29 +66,31 @@ def _is_connection_refused(exc: requests.ConnectionError) -> bool:
 
 def _handle_direct_put_transient_error(
     direct_put_url: str, exc: Exception, logger: logging.Logger
-) -> Tuple[int, bytes, int]:
+) -> Tuple[int, bytes, int, bool]:
     """
     Handle a caught SYNC_DIRECT_PUT_TRANSIENT_ERRORS exception.
 
-    Returns a ``(STATUS_BAD_GATEWAY, error_bytes, 0)`` tuple for permanent
-    ``ConnectionRefused`` errors.  Re-raises all other transient errors as
-    ``ETLDirectPutTransientError`` so the caller's retry loop can act on them.
+    Returns a ``(STATUS_BAD_GATEWAY, error_bytes, 0, False)`` tuple for
+    permanent ``ConnectionRefused`` errors.  Re-raises all other transient
+    errors as ``ETLDirectPutTransientError`` so the caller's retry loop can
+    act on them.
 
     Args:
         direct_put_url: The direct-put URL that was being contacted.
         exc: The caught exception (one of ``SYNC_DIRECT_PUT_TRANSIENT_ERRORS``).
         logger: Logger used to emit the permanent-error message.
 
     Returns:
-        ``(STATUS_BAD_GATEWAY, encoded_error_message, 0)`` for permanent errors.
+        ``(STATUS_BAD_GATEWAY, encoded_error_message, 0, False)`` for
+        permanent errors.
 
     Raises:
         ETLDirectPutTransientError: For all other transient errors.
     """
     if isinstance(exc, requests.ConnectionError) and _is_connection_refused(exc):
         error = f"direct_put to {direct_put_url!r} failed: {type(exc).__name__}: {exc}".encode()
         logger.error("Permanent connection error to %s: %s", direct_put_url, exc)
-        return STATUS_BAD_GATEWAY, error, 0
+        return STATUS_BAD_GATEWAY, error, 0, False
     raise ETLDirectPutTransientError(direct_put_url, exc) from exc
 
 
@@ -259,8 +263,17 @@ def iter_and_close(output_iter: Iterator[bytes], reader) -> Iterator[bytes]:
             ETLServer.close_reader(reader)
 
     @staticmethod
-    def make_direct_put_headers(direct_put_length: int) -> dict:
-        """Build response headers for a direct-put result."""
+    def make_direct_put_headers(direct_put_length: int, complete: bool = False) -> dict:
+        """Build response headers for a direct-put result.
+
+        When `complete` (the delivered ack carried `HEADER_DIRECT_PUT_COMPLETE`),
+        propagate the marker and the length verbatim, including a length of 0.
+        """
+        if complete:
+            return {
+                HEADER_DIRECT_PUT_COMPLETE: "true",
+                HEADER_DIRECT_PUT_LENGTH: str(direct_put_length),
+            }
         if direct_put_length != 0:
             return {HEADER_DIRECT_PUT_LENGTH: str(direct_put_length)}
         return {}
@@ -293,31 +306,62 @@ def client_put(
 
     def handle_direct_put_response(
         self, resp: requests.Response, data: bytes, data_length: int = -1
-    ) -> Tuple[int, bytes, int]:
+    ) -> Tuple[int, bytes, int, bool]:
         """Handle the response from a direct PUT request.
 
+        Returns a `(status, body, direct_put_length, direct_put_complete)`
+        tuple. `direct_put_complete` is True only when the response carried
+        `HEADER_DIRECT_PUT_COMPLETE` — the target's ack that it stored the
+        object — and must be propagated (see `make_direct_put_headers`).
+
         Args:
             resp: The HTTP response from the direct PUT.
             data: The original data bytes (used to compute length for the
-                200-OK-empty-content case). Can be `b""` for streaming.
+                legacy 200 + `Content-Length: 0` delivered case). Can be
+                `b""` for streaming.
             data_length: Explicit byte count override. When >= 0, used instead
                 of `len(data)`. Pass this from a `CountingIterator` for
                 streaming pipeline PUTs where `data` is empty.
         """
         size = data_length if data_length >= 0 else len(data)
 
+        # Delivered ack from the target (directly or propagated by a
+        # downstream stage). Presence-based: the value is ignored, and the
+        # marker decides regardless of status (the target pairs it with 204).
+        if HEADER_DIRECT_PUT_COMPLETE in resp.headers:
+            return (
+                STATUS_NO_CONTENT,
+                b"",
+                int(resp.headers.get(HEADER_DIRECT_PUT_LENGTH, "0")),
+                True,
+            )
+
+        # Legacy handling below, unchanged: kept for targets that predate
+        # HEADER_DIRECT_PUT_COMPLETE; to be phased out with them.
         if resp.status_code == STATUS_NO_CONTENT:
             return (
                 resp.status_code,
                 b"",
                 int(resp.headers.get(HEADER_DIRECT_PUT_LENGTH, "0")),
+                False,
             )
 
         if resp.status_code == STATUS_OK:
-            if resp.content:  # from other ETL server, forward the content back
-                return resp.status_code, resp.content, 0
+            # Keyed on the Content-Length header, mirroring the Go webserver's
+            # directPut (ext/etl/webserver/webserver.go): `0` means the next
+            # hop was the target — delivered, no content. Absent (chunked) or
+            # > 0 means transformed content from another ETL server; forward
+            # it as-is — an empty chunked body is a valid empty object.
+            content_length = resp.headers.get(HEADER_CONTENT_LENGTH)
+            try:
+                delivered = content_length is not None and int(content_length) == 0
+            except (TypeError, ValueError):
+                delivered = False  # malformed header: treat as content
+            if delivered:
+                return STATUS_NO_CONTENT, b"", size, False  # from target, no content
 
-            return STATUS_NO_CONTENT, b"", size  # from target, no content
+            # from other ETL server, forward the content back
+            return resp.status_code, resp.content, 0, False
 
         error = resp.content
         self.logger.error(
@@ -326,7 +370,7 @@ def handle_direct_put_response(
             resp.status_code,
             error,
         )
-        return resp.status_code, error, 0
+        return resp.status_code, error, 0, False
 
 
 class CountingIterator:  # pylint: disable=too-few-public-methods

python/aistore/sdk/etl/webserver/fastapi_server.py

@@ -50,7 +50,6 @@
     ETL_WS_FQN,
     ETL_WS_PATH,
     ETL_WS_PIPELINE,
-    HEADER_DIRECT_PUT_LENGTH,
     HEADER_ETL_RETRY_REASON,
     ETL_RETRY_REASON_DIRECT_PUT_TRANSIENT,
     QPARAM_ETL_ARGS,
@@ -268,7 +267,7 @@ async def _handle_request_buffered(self, path: str, request: Request, is_get: bo
         if pipeline_header:
             first_url, remaining_pipeline = parse_etl_pipeline(pipeline_header)
             if first_url:
-                status_code, transformed, direct_put_length = (
+                status_code, transformed, direct_put_length, direct_put_complete = (
                     await self._direct_put_with_retry(
                         first_url, transformed, remaining_pipeline, path, etl_args
                     )
@@ -278,10 +277,8 @@ async def _handle_request_buffered(self, path: str, request: Request, is_get: bo
                 return Response(
                     content=transformed,
                     status_code=status_code,
-                    headers=(
-                        {HEADER_DIRECT_PUT_LENGTH: str(direct_put_length)}
-                        if direct_put_length != 0
-                        else {}
+                    headers=self.make_direct_put_headers(
+                        direct_put_length, direct_put_complete
                     ),
                 )
 
@@ -314,7 +311,7 @@ async def _handle_request_streaming(
                 return Response(
                     content=result[1],
                     status_code=result[0],
-                    headers=self.make_direct_put_headers(result[2]),
+                    headers=self.make_direct_put_headers(result[2], result[3]),
                 )
 
         reader = await self._get_stream_reader(fqn, path, request, is_get)
@@ -374,7 +371,7 @@ async def _direct_put_stream_with_retry(  # pylint: disable=too-many-arguments,t
         etl_args: str,
         first_url: str,
         remaining: str,
-    ) -> Tuple[int, bytes, int]:
+    ) -> Tuple[int, bytes, int, bool]:
         """
         Stream-put with exponential-backoff retry on transient network errors.
 
@@ -401,8 +398,8 @@ async def _direct_put_stream_with_retry(  # pylint: disable=too-many-arguments,t
                 forwarded to the next stage via the `AIS-Node-Url` header.
 
         Returns:
-            Tuple[int, bytes, int]: `(status_code, body, length)` — see
-                `_direct_put_stream` for semantics.
+            Tuple[int, bytes, int, bool]: `(status_code, body, length,
+                complete)` — see `_direct_put_stream` for semantics.
 
         Raises:
             ETLDirectPutTransientError: if all retry attempts are exhausted.
@@ -458,15 +455,16 @@ async def _direct_put_stream(  # pylint: disable=too-many-arguments,too-many-pos
         remaining_pipeline: str = "",
         path: str = "",
         etl_args: str = "",
-    ) -> Tuple[int, bytes, int]:
+    ) -> Tuple[int, bytes, int, bool]:
         """
         Stream transformed output directly to the next pipeline stage.
 
         Returns:
-            (status_code, body, length) where:
+            (status_code, body, length, complete) where:
               - status_code: HTTP status of the PUT (200/204 on success, 500 on error).
               - body: response bytes forwarded back to the AIS target (empty on success).
               - length: bytes sent to the destination, from CountingIterator.
+              - complete: the ack carried HEADER_DIRECT_PUT_COMPLETE; propagate it.
         """
         try:
             url = compose_etl_direct_put_url(
@@ -493,7 +491,7 @@ async def _direct_put_stream(  # pylint: disable=too-many-arguments,too-many-pos
                 exc,
                 exc_info=True,
             )
-            return STATUS_INTERNAL_SERVER_ERROR, repr(exc).encode(), 0
+            return STATUS_INTERNAL_SERVER_ERROR, repr(exc).encode(), 0, False
 
     async def _get_fqn_content(self, path: str) -> bytes:
         """Safely read local file content with path normalization."""
@@ -531,12 +529,12 @@ async def _direct_put_with_retry(  # pylint: disable=too-many-arguments,too-many
         remaining_pipeline: str = "",
         path: str = "",
         etl_args: str = "",
-    ) -> Tuple[int, bytes, int]:
+    ) -> Tuple[int, bytes, int, bool]:
         """
         Buffered direct-put with exponential-backoff retry on transient network errors.
 
         Returns:
-            (status_code, body, length) — see _direct_put for semantics.
+            (status_code, body, length, complete) — see _direct_put for semantics.
         Raises:
             ETLDirectPutTransientError: if all retry attempts are exhausted.
         """
@@ -565,7 +563,7 @@ async def _direct_put(  # pylint: disable=too-many-arguments,too-many-positional
         remaining_pipeline: str = "",
         path: str = "",
         etl_args: str = "",
-    ) -> Tuple[int, bytes, int]:
+    ) -> Tuple[int, bytes, int, bool]:
         """
         Sends the transformed object directly to the specified AIS node (`direct_put_url`),
         eliminating the additional network hop through the original target.
@@ -578,7 +576,8 @@ async def _direct_put(  # pylint: disable=too-many-arguments,too-many-positional
             path: The path of the object.
             etl_args: Per-request transform arguments to forward to the next stage.
         Returns:
-            status code, transformed data, length of the transformed data (if any)
+            status code, transformed data, length of the transformed data (if any),
+            and whether the ack carried HEADER_DIRECT_PUT_COMPLETE
         Raises:
             ETLDirectPutTransientError: on ReadError/ConnectError/RemoteProtocolError
                 so the caller can retry without re-fetching data.
@@ -606,7 +605,7 @@ async def _direct_put(  # pylint: disable=too-many-arguments,too-many-positional
                 exc,
                 exc_info=True,
             )
-            return STATUS_INTERNAL_SERVER_ERROR, repr(exc).encode(), 0
+            return STATUS_INTERNAL_SERVER_ERROR, repr(exc).encode(), 0, False
 
     def _build_response(self, content: bytes, mime_type: str) -> Response:
         """Construct standardized response with appropriate headers."""
@@ -648,7 +647,7 @@ async def _handle_ws_message(self, websocket: WebSocket):
                 self.logger.debug("pipeline_header: %r", pipeline_header)
                 first_url, remaining_pipeline = parse_etl_pipeline(pipeline_header)
                 if first_url:
-                    status_code, transformed, direct_put_length = (
+                    status_code, transformed, direct_put_length, _ = (
                         await self._direct_put_with_retry(
                             first_url, transformed, remaining_pipeline, path, etl_args
                         )