feat(antd): byte-range downloads on the data streams (REST, gRPC, antd-go)

antd-go/client.go

@@ -139,6 +139,27 @@ func (c *Client) doStream(ctx context.Context, method, path string, body any) (i
 // "application/x-ndjson" opts the stream endpoints into interleaved NDJSON
 // progress framing; an empty accept leaves the default raw octet-stream body.
 func (c *Client) doStreamWithAccept(ctx context.Context, method, path string, body any, accept string) (io.ReadCloser, error) {
+	header := http.Header{}
+	if accept != "" {
+		header.Set("Accept", accept)
+	}
+	resp, err := c.openStream(ctx, method, path, body, header)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		defer resp.Body.Close()
+		respBytes, _ := io.ReadAll(resp.Body)
+		return nil, errorFromBody(resp.StatusCode, respBytes)
+	}
+
+	return resp.Body, nil
+}
+
+// openStream issues a request with the given extra headers and returns the
+// response, whatever its status, with the body still open for the caller.
+func (c *Client) openStream(ctx context.Context, method, path string, body any, header http.Header) (*http.Response, error) {
 	var reqBody io.Reader
 	if body != nil {
 		b, err := json.Marshal(body)
@@ -155,22 +176,53 @@ func (c *Client) doStreamWithAccept(ctx context.Context, method, path string, bo
 	if body != nil {
 		req.Header.Set("Content-Type", "application/json")
 	}
-	if accept != "" {
-		req.Header.Set("Accept", accept)
+	for name, values := range header {
+		req.Header[name] = values
 	}
 
 	resp, err := c.http.Do(req)
 	if err != nil {
 		return nil, fmt.Errorf("http request: %w", err)
 	}
+	return resp, nil
+}
 
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+// doRangeStream issues a streaming request with a Range header and returns the
+// 206 body as a RangeReader. A 416 becomes a *RangeNotSatisfiableError carrying
+// the object size; a 200 means the daemon ignored the range.
+func (c *Client) doRangeStream(ctx context.Context, method, path string, body any, offset, length int64) (*RangeReader, error) {
+	value, err := rangeHeader(offset, length)
+	if err != nil {
+		return nil, err
+	}
+	header := http.Header{}
+	header.Set("Range", value)
+	resp, err := c.openStream(ctx, method, path, body, header)
+	if err != nil {
+		return nil, err
+	}
+
+	switch {
+	case resp.StatusCode == http.StatusPartialContent:
+		r, err := parseContentRange(resp.Header.Get("Content-Range"))
+		if err != nil {
+			resp.Body.Close()
+			return nil, err
+		}
+		r.ReadCloser = resp.Body
+		return r, nil
+	case resp.StatusCode >= 200 && resp.StatusCode < 300:
+		resp.Body.Close()
+		return nil, errNoRangeSupport
+	default:
 		defer resp.Body.Close()
 		respBytes, _ := io.ReadAll(resp.Body)
-		return nil, errorFromBody(resp.StatusCode, respBytes)
+		err := errorFromBody(resp.StatusCode, respBytes)
+		if rerr, ok := err.(*RangeNotSatisfiableError); ok {
+			rerr.Size = unsatisfiedSize(resp.Header.Get("Content-Range"))
+		}
+		return nil, err
 	}
-
-	return resp.Body, nil
 }
 
 func str(m map[string]any, key string) string {
@@ -315,6 +367,24 @@ func (c *Client) DataStreamPublic(ctx context.Context, address string) (io.ReadC
 	return c.doStream(ctx, http.MethodGet, "/v1/data/public/"+address+"/stream", nil)
 }
 
+// DataStreamRange streams the plaintext bytes [offset, offset+length) of
+// private data from a caller-held DataMap (hex), fetching only the chunks that
+// overlap them. Pass RangeToEnd as length to read to the end; a length past
+// the end is clamped. An offset at or past the end is a
+// *RangeNotSatisfiableError. The caller reads the returned stream and MUST
+// Close it.
+func (c *Client) DataStreamRange(ctx context.Context, dataMap string, offset, length int64) (*RangeReader, error) {
+	return c.doRangeStream(ctx, http.MethodPost, "/v1/data/stream", map[string]any{
+		"data_map": dataMap,
+	}, offset, length)
+}
+
+// DataStreamPublicRange streams a byte range of public data by address — the
+// public counterpart to DataStreamRange, with the same semantics.
+func (c *Client) DataStreamPublicRange(ctx context.Context, address string, offset, length int64) (*RangeReader, error) {
+	return c.doRangeStream(ctx, http.MethodGet, "/v1/data/public/"+address+"/stream", nil, offset, length)
+}
+
 // ndjsonContentType opts the stream endpoints into NDJSON progress framing.
 const ndjsonContentType = "application/x-ndjson"
 

antd-go/errors.go

@@ -49,6 +49,15 @@ type NetworkError struct{ AntdError }
 // dependency such as a wallet (HTTP 503).
 type ServiceUnavailableError struct{ AntdError }
 
+// RangeNotSatisfiableError indicates a byte-range download that starts at or
+// past the end of the object (HTTP 416; gRPC OUT_OF_RANGE). Size is the
+// object's plaintext size when the daemon reported it (REST, from the 416's
+// Content-Range), or -1 when unknown (gRPC).
+type RangeNotSatisfiableError struct {
+	AntdError
+	Size int64
+}
+
 // PartialUploadError indicates a finalize stored some chunks while others
 // remained unstored after the daemon's retries (HTTP 502 with code
 // PARTIAL_UPLOAD; gRPC ABORTED). The on-chain payment persists and the
@@ -287,6 +296,8 @@ func errorForStatus(statusCode int, message string) error {
 		return &AlreadyExistsError{base}
 	case 413:
 		return &TooLargeError{base}
+	case 416:
+		return &RangeNotSatisfiableError{AntdError: base, Size: -1}
 	case 500:
 		return &InternalError{base}
 	case 502:

antd-go/grpc_client.go

@@ -168,6 +168,9 @@ func errorFromGrpc(err error) error {
 	case codes.ResourceExhausted:
 		base.StatusCode = 413
 		return &TooLargeError{base}
+	case codes.OutOfRange:
+		base.StatusCode = 416
+		return &RangeNotSatisfiableError{AntdError: base, Size: -1}
 	case codes.Internal:
 		base.StatusCode = 500
 		return &InternalError{base}
@@ -361,6 +364,87 @@ func (c *GrpcClient) DataStreamPublic(ctx context.Context, address string) (io.R
 	return &grpcChunkReader{stream: stream, cancel: cancel}, nil
 }
 
+// DataStreamRange streams the plaintext bytes [offset, offset+length) of
+// private data from a caller-held DataMap (hex), fetching only the chunks that
+// overlap them — the gRPC mirror of the REST client's DataStreamRange. Pass
+// RangeToEnd as length to read to the end; a length past the end is clamped.
+// An offset at or past the end is a *RangeNotSatisfiableError. The caller
+// MUST Close the returned reader (Close cancels the RPC).
+func (c *GrpcClient) DataStreamRange(ctx context.Context, dataMap string, offset, length int64) (*RangeReader, error) {
+	if err := checkRange(offset, length); err != nil {
+		return nil, err
+	}
+	req := &pb.StreamDataRequest{DataMap: dataMap}
+	req.Offset, req.Length = grpcRange(offset, length)
+	ctx, cancel := c.ctx(ctx)
+	stream, err := c.data.Stream(ctx, req)
+	if err != nil {
+		cancel()
+		return nil, errorFromGrpc(err)
+	}
+	return openGrpcRange(stream, cancel)
+}
+
+// DataStreamPublicRange streams a byte range of public data by address — the
+// public counterpart to DataStreamRange, with the same semantics.
+func (c *GrpcClient) DataStreamPublicRange(ctx context.Context, address string, offset, length int64) (*RangeReader, error) {
+	if err := checkRange(offset, length); err != nil {
+		return nil, err
+	}
+	req := &pb.StreamPublicDataRequest{Address: address}
+	req.Offset, req.Length = grpcRange(offset, length)
+	ctx, cancel := c.ctx(ctx)
+	stream, err := c.data.StreamPublic(ctx, req)
+	if err != nil {
+		cancel()
+		return nil, errorFromGrpc(err)
+	}
+	return openGrpcRange(stream, cancel)
+}
+
+// grpcRange converts a validated offset/length into the request's optional
+// fields; RangeToEnd leaves length unset.
+func grpcRange(offset, length int64) (*uint64, *uint64) {
+	o := uint64(offset)
+	if length == RangeToEnd {
+		return &o, nil
+	}
+	l := uint64(length)
+	return &o, &l
+}
+
+// openGrpcRange waits for a ranged stream's response headers and reads the
+// range from x-content-range. A stream that failed before sending headers
+// (e.g. OUT_OF_RANGE) surfaces its status here rather than on the first Read.
+// A daemon that predates range support ignores the fields and streams the whole
+// object without x-content-range, which is reported as an error rather than
+// passed off as the range.
+func openGrpcRange(stream dataChunkStream, cancel context.CancelFunc) (*RangeReader, error) {
+	md, err := stream.Header()
+	if err != nil {
+		cancel()
+		return nil, errorFromGrpc(err)
+	}
+	values := md.Get("x-content-range")
+	if len(values) == 0 {
+		// No x-content-range: either the RPC ended trailers-only (its status
+		// comes from Recv) or an old daemon is streaming the whole object.
+		_, err := stream.Recv()
+		cancel()
+		if err != nil && err != io.EOF {
+			return nil, errorFromGrpc(err)
+		}
+		return nil, errNoRangeSupport
+	}
+	r, err := parseContentRange(values[0])
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+	r.ReadCloser = &grpcChunkReader{stream: stream, cancel: cancel}
+	return r, nil
+}
+
 // DownloadStream is a progress-enabled streaming download: a sequence of
 // [DownloadFrame]s, each either a plaintext data chunk or a [DownloadProgress]
 // update interleaved on the same RPC. Returned by DataStreamWithProgress /

antd-go/grpc_client_test.go

@@ -81,6 +81,9 @@ func (m *mockDataService) Get(_ context.Context, _ *pb.GetDataRequest) (*pb.GetD
 // Stream emits the private payload as two chunks so the reader's
 // chunk-boundary buffering is exercised, not just a single Recv.
 func (m *mockDataService) Stream(req *pb.StreamDataRequest, srv grpc.ServerStreamingServer[pb.DataChunk]) error {
+	if ranged, err := serveMockRange("secret", req.GetDataMap(), req.Offset, req.Length, srv); ranged {
+		return err
+	}
 	if req.GetIncludeProgress() {
 		// Mirror the daemon: attach the byte total as response metadata.
 		_ = srv.SetHeader(metadata.Pairs("x-content-length", "6"))
@@ -99,6 +102,9 @@ func (m *mockDataService) Stream(req *pb.StreamDataRequest, srv grpc.ServerStrea
 }
 
 func (m *mockDataService) StreamPublic(req *pb.StreamPublicDataRequest, srv grpc.ServerStreamingServer[pb.DataChunk]) error {
+	if ranged, err := serveMockRange("hello", req.GetAddress(), req.Offset, req.Length, srv); ranged {
+		return err
+	}
 	if req.GetIncludeProgress() {
 		_ = srv.SetHeader(metadata.Pairs("x-content-length", "5"))
 		if err := srv.Send(&pb.DataChunk{Kind: &pb.DataChunk_Progress{Progress: &pb.DownloadProgress{