etl: acknowledge completed direct puts

ais/tgtetl.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/url"
+	"strconv"
 	"strings"
 
 	"github.com/NVIDIA/aistore/api/apc"
@@ -329,6 +330,9 @@ func (t *target) getObjectETL(w http.ResponseWriter, r *http.Request, dpq *dpq,
 
 // PUT /v1/etl/_object/<etl-name>/<secret>/<uname>?uuid=<xid>
 // Handles PUT requests from ETL containers (K8s Pods).
+// Acknowledges a successful direct put with 204 (No Content) and the
+// (apc.HdrDirectPutComplete, apc.HdrDirectPutLength) response headers;
+// ETL webservers propagate this ack back through the pipeline.
 func (t *target) putObjectETL(w http.ResponseWriter, r *http.Request, dpq *dpq, bck *meta.Bck, objName string) {
 	config := cmn.GCO.Get()
 	lom := core.AllocLOM(objName)
@@ -344,10 +348,17 @@ func (t *target) putObjectETL(w http.ResponseWriter, r *http.Request, dpq *dpq,
 		}
 	}
 	ecode, err := t.putObject(w, r, dpq, lom, config)
-	core.FreeLOM(lom)
 	if err != nil {
+		core.FreeLOM(lom)
 		t.writeErr(w, r, err, ecode)
+		return
 	}
+	size := lom.Lsize()
+	core.FreeLOM(lom)
+
+	w.Header().Set(apc.HdrDirectPutComplete, "true")
+	w.Header().Set(apc.HdrDirectPutLength, strconv.FormatInt(size, 10))
+	w.WriteHeader(http.StatusNoContent)
 }
 
 // HEAD /v1/etl/_object/<etl-name>/<secret>/<uname>

api/apc/headers.go

@@ -131,6 +131,10 @@ const (
 	// ETL
 	HdrETLPodInfo      = aisPrefix + "ETL-Pod-Info" // serialized etl.Info
 	HdrDirectPutLength = aisPrefix + "Direct-Put-Length"
+	// Set by the target (alongside 204 and HdrDirectPutLength) after storing an
+	// object via the direct-put endpoint. Presence-based - the value is ignored;
+	// intermediate ETL webservers propagate it back through the pipeline.
+	HdrDirectPutComplete = aisPrefix + "Direct-Put-Complete"
 	// ETL → AIS retry contract: emitted by the ETL webserver alongside HTTP 503
 	// to signal that the ETL bailed on a transient direct-put failure without
 	// trying locally (one-shot body case). AIS retries the whole PUT against

ext/etl/webserver/webserver.go

@@ -79,6 +79,7 @@ type (
 		StatusCode      int
 		Size            int64
 		DirectPutLength int64
+		Complete        bool // ack carried apc.HdrDirectPutComplete; propagate it
 		Body            io.ReadCloser
 	}
 )
@@ -310,7 +311,12 @@ func (base *etlServerBase) handleDirectPut(w http.ResponseWriter, transformedRea
 	if dresp.Body != nil {
 		setResponseHeaders(w.Header(), dresp.Size)
 	}
-	if dresp.DirectPutLength > 0 {
+	if dresp.Complete {
+		// delivered ack: propagate the marker and the length verbatim,
+		// including a length of 0 (empty object stored)
+		w.Header().Set(apc.HdrDirectPutComplete, "true")
+		w.Header().Set(apc.HdrDirectPutLength, strconv.FormatInt(dresp.DirectPutLength, 10))
+	} else if dresp.DirectPutLength > 0 {
 		w.Header().Set(apc.HdrDirectPutLength, strconv.FormatInt(dresp.DirectPutLength, 10))
 	}
 	w.WriteHeader(dresp.StatusCode)
@@ -447,6 +453,22 @@ func (base *etlServerBase) directPut(directPutURL string, r io.ReadCloser, size
 		length = directPutLength
 	}
 
+	// 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 _, ok := resp.Header[apc.HdrDirectPutComplete]; ok {
+		resp.Body.Close()
+		return &directPutResponse{
+			StatusCode:      http.StatusNoContent,
+			DirectPutLength: int64(length),
+			Complete:        true,
+			Body:            nil,
+		}, nil
+	}
+
+	// Legacy handling below, unchanged: kept for targets that predate
+	// apc.HdrDirectPutComplete; to be phased out with them.
+
 	// delivered to target, no content
 	if resp.StatusCode == http.StatusNoContent {
 		resp.Body.Close()
@@ -458,11 +480,9 @@ func (base *etlServerBase) directPut(directPutURL string, r io.ReadCloser, size
 	}
 
 	if resp.StatusCode == http.StatusOK {
-		// NOTE: keyed on Content-Length, not the read body (contrast Python's
-		// handle_direct_put_response): a chunked 200 (ContentLength == -1) is
-		// always forwarded as content - an empty transform result is a valid
-		// object. Python currently maps any empty 200 body to 204/"delivered";
-		// to be aligned with this behavior in a separate PR.
+		// NOTE: keyed on Content-Length, not the read body: a chunked 200
+		// (ContentLength == -1) is always forwarded as content because an
+		// empty transform result is a valid object.
 
 		// from target, no content
 		if resp.ContentLength == 0 {

ext/etl/webserver/webserver_internal_test.go

@@ -282,6 +282,117 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) {
 		directPutPath = "ais@#test/obj"
 	)
 
+	t.Run("204 with Direct-Put-Complete propagates marker and length", func(t *testing.T) {
+		const directPutLength = 1234
+		directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method)
+			w.Header().Set(apc.HdrDirectPutComplete, "true")
+			w.Header().Set(apc.HdrDirectPutLength, strconv.Itoa(directPutLength))
+			w.WriteHeader(http.StatusNoContent)
+		}))
+		defer directPutTargetServer.Close()
+
+		var (
+			content = []byte("test bytes")
+			req     = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content))
+			w       = httptest.NewRecorder()
+		)
+		req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}}
+
+		svr.putHandler(w, req)
+
+		resp := w.Result()
+		defer resp.Body.Close()
+
+		tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode)
+		_, marked := resp.Header[apc.HdrDirectPutComplete]
+		tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated")
+		got := resp.Header.Get(apc.HdrDirectPutLength)
+		tassert.Fatalf(t, got == strconv.Itoa(directPutLength), "expected Direct-Put-Length %d, got %q", directPutLength, got)
+	})
+
+	t.Run("Direct-Put-Complete with length 0 propagates 0", func(t *testing.T) {
+		directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method)
+			w.Header().Set(apc.HdrDirectPutComplete, "true")
+			w.Header().Set(apc.HdrDirectPutLength, "0") // empty object stored
+			w.WriteHeader(http.StatusNoContent)
+		}))
+		defer directPutTargetServer.Close()
+
+		var (
+			content = []byte("test bytes")
+			req     = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content))
+			w       = httptest.NewRecorder()
+		)
+		req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}}
+
+		svr.putHandler(w, req)
+
+		resp := w.Result()
+		defer resp.Body.Close()
+
+		tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode)
+		_, marked := resp.Header[apc.HdrDirectPutComplete]
+		tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated")
+		got := resp.Header.Get(apc.HdrDirectPutLength)
+		tassert.Fatalf(t, got == "0", "expected Direct-Put-Length 0, got %q", got)
+	})
+
+	t.Run("Direct-Put-Complete with empty value still delivered", func(t *testing.T) {
+		directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method)
+			w.Header().Set(apc.HdrDirectPutComplete, "") // presence-based: value is ignored
+			w.WriteHeader(http.StatusNoContent)
+		}))
+		defer directPutTargetServer.Close()
+
+		var (
+			content = []byte("test bytes")
+			req     = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content))
+			w       = httptest.NewRecorder()
+		)
+		req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}}
+
+		svr.putHandler(w, req)
+
+		resp := w.Result()
+		defer resp.Body.Close()
+
+		tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode)
+		_, marked := resp.Header[apc.HdrDirectPutComplete]
+		tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated")
+	})
+
+	t.Run("marked 200 normalized to 204", func(t *testing.T) {
+		directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method)
+			w.Header().Set(apc.HdrDirectPutComplete, "true")
+			w.Header().Set(apc.HdrDirectPutLength, "5")
+			w.WriteHeader(http.StatusOK) // the marker decides regardless of status
+		}))
+		defer directPutTargetServer.Close()
+
+		var (
+			content = []byte("test bytes")
+			req     = httptest.NewRequest(http.MethodPut, "/", bytes.NewReader(content))
+			w       = httptest.NewRecorder()
+		)
+		req.Header = http.Header{apc.HdrNodeURL: []string{cos.JoinPath(directPutTargetServer.URL, url.PathEscape(directPutPath))}}
+
+		svr.putHandler(w, req)
+
+		resp := w.Result()
+		defer resp.Body.Close()
+
+		tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode)
+		_, marked := resp.Header[apc.HdrDirectPutComplete]
+		tassert.Fatalf(t, marked, "expected Direct-Put-Complete to be propagated")
+		got := resp.Header.Get(apc.HdrDirectPutLength)
+		tassert.Fatalf(t, got == "5", "expected Direct-Put-Length 5, got %q", got)
+	})
+
+	// legacy path: bare 204 ack from a target that predates Direct-Put-Complete
 	t.Run("204 forwards Direct-Put-Length", func(t *testing.T) {
 		const directPutLength = 1234
 		directPutTargetServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -304,6 +415,8 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) {
 		defer resp.Body.Close()
 
 		tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode)
+		_, marked := resp.Header[apc.HdrDirectPutComplete]
+		tassert.Fatalf(t, !marked, "legacy delivery must not gain Direct-Put-Complete")
 		got := resp.Header.Get(apc.HdrDirectPutLength)
 		tassert.Fatalf(t, got == strconv.Itoa(directPutLength), "expected Direct-Put-Length %d, got %q", directPutLength, got)
 	})
@@ -368,6 +481,7 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) {
 		tassert.Fatalf(t, bytes.Equal(result, expected), "expected body %s, got %s", expected, result)
 	})
 
+	// legacy path: bare 200 ack from a target that predates Direct-Put-Complete
 	t.Run("200 empty body maps to 204", func(t *testing.T) {
 		nextStageServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			tassert.Fatalf(t, r.Method == http.MethodPut, "expected PUT method, got %s", r.Method)
@@ -388,6 +502,8 @@ func TestETLServerDirectPutResponseForwarding(t *testing.T) {
 		defer resp.Body.Close()
 
 		tassert.Fatalf(t, http.StatusNoContent == resp.StatusCode, "expected status code 204, got %d", resp.StatusCode)
+		_, marked := resp.Header[apc.HdrDirectPutComplete]
+		tassert.Fatalf(t, !marked, "legacy delivery must not gain Direct-Put-Complete")
 		got := resp.Header.Get(apc.HdrDirectPutLength)
 		tassert.Fatalf(t, got == strconv.Itoa(len(content)), "expected Direct-Put-Length %d, got %q", len(content), got)
 	})