From f2b0bd30b19bbad51c52f783757617e0966f0a55 Mon Sep 17 00:00:00 2001 From: Nic Dorman Date: Thu, 8 Oct 2026 11:00:16 +0000 Subject: [PATCH] feat(antd): byte-range downloads on the data streams (REST, gRPC, antd-go) antd could only stream whole objects. ant-core already reads byte ranges (Client::data_download_range fetches and decrypts only the chunks a range overlaps); this exposes that. - REST: one `Range: bytes=` range on POST /v1/data/stream and GET /v1/data/public/{addr}/stream -> 206 + Content-Range, 416 + `bytes */size` (RANGE_NOT_SATISFIABLE), 400 for malformed or multiple ranges or Range with NDJSON progress. Whole-object responses send Accept-Ranges: bytes. - gRPC: optional offset/length on both stream requests; OUT_OF_RANGE past the end, INVALID_ARGUMENT for length 0 or a range with include_progress; x-content-length / x-content-range metadata. - antd/src/range.rs reads a range in chunk-aligned windows of 8 chunks, so memory stays bounded and every chunk is fetched once. - antd-go: DataStreamRange / DataStreamPublicRange on both clients, returning a *RangeReader stream; RangeToEnd, RangeNotSatisfiableError. - OpenAPI documents Range/206/416 and the missing POST /v1/data/stream. - Regenerated Go and Swift stubs, synced the Kotlin proto copy, updated antd-rust's generated code. Closes V2-1425 Co-Authored-By: Claude Opus 5.5 --- antd-go/client.go | 82 +++- antd-go/errors.go | 11 + antd-go/grpc_client.go | 84 +++++ antd-go/grpc_client_test.go | 6 + antd-go/proto/antd/v1/data.pb.go | 71 +++- antd-go/proto/antd/v1/health.pb.go | 75 +++- antd-go/proto/antd/v1/upload.pb.go | 23 +- antd-go/proto/antd/v1/upload_grpc.pb.go | 2 +- antd-go/range.go | 98 +++++ antd-go/range_test.go | 206 ++++++++++ .../lib/src/main/proto/antd/v1/data.proto | 13 + antd-rust/src/generated/antd.v1.rs | 17 + antd-rust/src/grpc_client.rs | 4 + .../AntdSdk/Proto/antd/v1/data.pb.swift | 83 +++- antd/openapi.yaml | 96 ++++- antd/proto/antd/v1/data.proto | 13 + antd/src/error.rs | 41 +- antd/src/grpc/service.rs | 102 +++++ antd/src/main.rs | 1 + antd/src/range.rs | 357 ++++++++++++++++++ antd/src/rest/data.rs | 74 +++- 21 files changed, 1415 insertions(+), 44 deletions(-) create mode 100644 antd-go/range.go create mode 100644 antd-go/range_test.go create mode 100644 antd/src/range.rs diff --git a/antd-go/client.go b/antd-go/client.go index f1bdf889..6a059a97 100644 --- a/antd-go/client.go +++ b/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" diff --git a/antd-go/errors.go b/antd-go/errors.go index d138404b..d8b7e210 100644 --- a/antd-go/errors.go +++ b/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: diff --git a/antd-go/grpc_client.go b/antd-go/grpc_client.go index 048bc802..63d0c933 100644 --- a/antd-go/grpc_client.go +++ b/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 / diff --git a/antd-go/grpc_client_test.go b/antd-go/grpc_client_test.go index ee12f01d..70059def 100644 --- a/antd-go/grpc_client_test.go +++ b/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{ diff --git a/antd-go/proto/antd/v1/data.pb.go b/antd-go/proto/antd/v1/data.pb.go index 65ca071a..c36d5f46 100644 --- a/antd-go/proto/antd/v1/data.pb.go +++ b/antd-go/proto/antd/v1/data.pb.go @@ -242,8 +242,18 @@ type StreamPublicDataRequest struct { // progress bar. Defaults to false — old clients omitting it receive a // pure data-frame stream, byte-identical to the pre-progress behaviour. IncludeProgress bool `protobuf:"varint,2,opt,name=include_progress,json=includeProgress,proto3" json:"include_progress,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Byte-range download. Setting either field streams only the plaintext bytes + // [offset, offset + length) instead of the whole object, fetching just the + // chunks that overlap them. `offset` defaults to 0 and an absent `length` + // reads to the end; a `length` past the end is clamped. `length` 0 is + // INVALID_ARGUMENT, and an `offset` at or past the end is OUT_OF_RANGE. A + // ranged stream's `x-content-length` metadata is the range's byte count, and + // it adds `x-content-range` (`bytes first-last/size`, `last` inclusive). + // Cannot be combined with include_progress (INVALID_ARGUMENT). + Offset *uint64 `protobuf:"varint,3,opt,name=offset,proto3,oneof" json:"offset,omitempty"` + Length *uint64 `protobuf:"varint,4,opt,name=length,proto3,oneof" json:"length,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *StreamPublicDataRequest) Reset() { @@ -290,6 +300,20 @@ func (x *StreamPublicDataRequest) GetIncludeProgress() bool { return false } +func (x *StreamPublicDataRequest) GetOffset() uint64 { + if x != nil && x.Offset != nil { + return *x.Offset + } + return 0 +} + +func (x *StreamPublicDataRequest) GetLength() uint64 { + if x != nil && x.Length != nil { + return *x.Length + } + return 0 +} + // A single frame of a streaming download. Exactly one of `data` (a decrypted // plaintext chunk) or `progress` (a fetch-progress update) is set per frame. // `data` keeps wire tag 1, so a frame carrying data is byte-identical to the @@ -385,7 +409,9 @@ func (*DataChunk_Progress) isDataChunk_Kind() {} // actually advance smoothly during a download. type DownloadProgress struct { state protoimpl.MessageState `protogen:"open.v1"` - // One of: "resolving_map", "resolved", "fetching". + // One of: "resolving_map", "resolved", "fetching". Daemons that resolve a + // shrunk DataMap before the stream opens (V2-1104) no longer emit + // "resolving_map" frames; the value remains for streams from older daemons. Phase string `protobuf:"bytes,1,opt,name=phase,proto3" json:"phase,omitempty"` // Chunks fetched so far in the current phase. Fetched uint64 `protobuf:"varint,2,opt,name=fetched,proto3" json:"fetched,omitempty"` @@ -495,8 +521,11 @@ type StreamDataRequest struct { DataMap string `protobuf:"bytes,1,opt,name=data_map,json=dataMap,proto3" json:"data_map,omitempty"` // hex // See StreamPublicDataRequest.include_progress. Defaults to false. IncludeProgress bool `protobuf:"varint,2,opt,name=include_progress,json=includeProgress,proto3" json:"include_progress,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Byte range: see StreamPublicDataRequest.offset / length. + Offset *uint64 `protobuf:"varint,3,opt,name=offset,proto3,oneof" json:"offset,omitempty"` + Length *uint64 `protobuf:"varint,4,opt,name=length,proto3,oneof" json:"length,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *StreamDataRequest) Reset() { @@ -543,6 +572,20 @@ func (x *StreamDataRequest) GetIncludeProgress() bool { return false } +func (x *StreamDataRequest) GetOffset() uint64 { + if x != nil && x.Offset != nil { + return *x.Offset + } + return 0 +} + +func (x *StreamDataRequest) GetLength() uint64 { + if x != nil && x.Length != nil { + return *x.Length + } + return 0 +} + type GetDataResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` @@ -782,10 +825,14 @@ const file_antd_v1_data_proto_rawDesc = "" + "\x04cost\x18\x01 \x01(\v2\r.antd.v1.CostR\x04cost\x12\x18\n" + "\aaddress\x18\x02 \x01(\tR\aaddress\x12#\n" + "\rchunks_stored\x18\x03 \x01(\x04R\fchunksStored\x12*\n" + - "\x11payment_mode_used\x18\x04 \x01(\tR\x0fpaymentModeUsed\"^\n" + + "\x11payment_mode_used\x18\x04 \x01(\tR\x0fpaymentModeUsed\"\xae\x01\n" + "\x17StreamPublicDataRequest\x12\x18\n" + "\aaddress\x18\x01 \x01(\tR\aaddress\x12)\n" + - "\x10include_progress\x18\x02 \x01(\bR\x0fincludeProgress\"b\n" + + "\x10include_progress\x18\x02 \x01(\bR\x0fincludeProgress\x12\x1b\n" + + "\x06offset\x18\x03 \x01(\x04H\x00R\x06offset\x88\x01\x01\x12\x1b\n" + + "\x06length\x18\x04 \x01(\x04H\x01R\x06length\x88\x01\x01B\t\n" + + "\a_offsetB\t\n" + + "\a_length\"b\n" + "\tDataChunk\x12\x14\n" + "\x04data\x18\x01 \x01(\fH\x00R\x04data\x127\n" + "\bprogress\x18\x02 \x01(\v2\x19.antd.v1.DownloadProgressH\x00R\bprogressB\x06\n" + @@ -795,10 +842,14 @@ const file_antd_v1_data_proto_rawDesc = "" + "\afetched\x18\x02 \x01(\x04R\afetched\x12\x14\n" + "\x05total\x18\x03 \x01(\x04R\x05total\"+\n" + "\x0eGetDataRequest\x12\x19\n" + - "\bdata_map\x18\x01 \x01(\tR\adataMap\"Y\n" + + "\bdata_map\x18\x01 \x01(\tR\adataMap\"\xa9\x01\n" + "\x11StreamDataRequest\x12\x19\n" + "\bdata_map\x18\x01 \x01(\tR\adataMap\x12)\n" + - "\x10include_progress\x18\x02 \x01(\bR\x0fincludeProgress\"%\n" + + "\x10include_progress\x18\x02 \x01(\bR\x0fincludeProgress\x12\x1b\n" + + "\x06offset\x18\x03 \x01(\x04H\x00R\x06offset\x88\x01\x01\x12\x1b\n" + + "\x06length\x18\x04 \x01(\x04H\x01R\x06length\x88\x01\x01B\t\n" + + "\a_offsetB\t\n" + + "\a_length\"%\n" + "\x0fGetDataResponse\x12\x12\n" + "\x04data\x18\x01 \x01(\fR\x04data\"G\n" + "\x0ePutDataRequest\x12\x12\n" + @@ -881,10 +932,12 @@ func file_antd_v1_data_proto_init() { return } file_antd_v1_common_proto_init() + file_antd_v1_data_proto_msgTypes[4].OneofWrappers = []any{} file_antd_v1_data_proto_msgTypes[5].OneofWrappers = []any{ (*DataChunk_Data)(nil), (*DataChunk_Progress)(nil), } + file_antd_v1_data_proto_msgTypes[8].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ diff --git a/antd-go/proto/antd/v1/health.pb.go b/antd-go/proto/antd/v1/health.pb.go index c99d5f73..0eec495f 100644 --- a/antd-go/proto/antd/v1/health.pb.go +++ b/antd-go/proto/antd/v1/health.pb.go @@ -58,17 +58,22 @@ func (*HealthCheckRequest) Descriptor() ([]byte, []int) { } type HealthCheckResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Status string `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"` // "ok" - Network string `protobuf:"bytes,2,opt,name=network,proto3" json:"network,omitempty"` // "default", "local", "alpha" - Version string `protobuf:"bytes,3,opt,name=version,proto3" json:"version,omitempty"` // antd crate version (e.g. "0.4.0") - EvmNetwork string `protobuf:"bytes,4,opt,name=evm_network,json=evmNetwork,proto3" json:"evm_network,omitempty"` // EVM preset: "arbitrum-one", "arbitrum-sepolia", "local", "custom" - UptimeSeconds uint64 `protobuf:"varint,5,opt,name=uptime_seconds,json=uptimeSeconds,proto3" json:"uptime_seconds,omitempty"` // seconds since process start - BuildCommit string `protobuf:"bytes,6,opt,name=build_commit,json=buildCommit,proto3" json:"build_commit,omitempty"` // short git SHA, or "" if built outside a git checkout - PaymentTokenAddress string `protobuf:"bytes,7,opt,name=payment_token_address,json=paymentTokenAddress,proto3" json:"payment_token_address,omitempty"` // token contract, or "" if unconfigured - PaymentVaultAddress string `protobuf:"bytes,8,opt,name=payment_vault_address,json=paymentVaultAddress,proto3" json:"payment_vault_address,omitempty"` // payment vault contract, or "" if unconfigured - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Status string `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"` // "ok" + Network string `protobuf:"bytes,2,opt,name=network,proto3" json:"network,omitempty"` // "default", "local", "alpha" + Version string `protobuf:"bytes,3,opt,name=version,proto3" json:"version,omitempty"` // antd crate version (e.g. "0.4.0") + EvmNetwork string `protobuf:"bytes,4,opt,name=evm_network,json=evmNetwork,proto3" json:"evm_network,omitempty"` // EVM preset: "arbitrum-one", "arbitrum-sepolia", "local", "custom" + UptimeSeconds uint64 `protobuf:"varint,5,opt,name=uptime_seconds,json=uptimeSeconds,proto3" json:"uptime_seconds,omitempty"` // seconds since process start + BuildCommit string `protobuf:"bytes,6,opt,name=build_commit,json=buildCommit,proto3" json:"build_commit,omitempty"` // short git SHA, or "" if built outside a git checkout + PaymentTokenAddress string `protobuf:"bytes,7,opt,name=payment_token_address,json=paymentTokenAddress,proto3" json:"payment_token_address,omitempty"` // token contract, or "" if unconfigured + PaymentVaultAddress string `protobuf:"bytes,8,opt,name=payment_vault_address,json=paymentVaultAddress,proto3" json:"payment_vault_address,omitempty"` // payment vault contract, or "" if unconfigured + WriteReady bool `protobuf:"varint,9,opt,name=write_ready,json=writeReady,proto3" json:"write_ready,omitempty"` // best-effort write-path signal: max(routing_table_size, connected_peers) >= rebootstrap_threshold; false = stores known-degraded + ConnectedPeers uint32 `protobuf:"varint,10,opt,name=connected_peers,json=connectedPeers,proto3" json:"connected_peers,omitempty"` // live transport-level connection count (distinct from routing_table_size) + RoutingTableSize uint32 `protobuf:"varint,11,opt,name=routing_table_size,json=routingTableSize,proto3" json:"routing_table_size,omitempty"` // DHT routing-table entries — the number auto-re-bootstrap keys off + RebootstrapThreshold uint32 `protobuf:"varint,12,opt,name=rebootstrap_threshold,json=rebootstrapThreshold,proto3" json:"rebootstrap_threshold,omitempty"` // routing-table floor below which the DHT auto-re-bootstraps + LastStoreOkSecsAgo *uint64 `protobuf:"varint,13,opt,name=last_store_ok_secs_ago,json=lastStoreOkSecsAgo,proto3,oneof" json:"last_store_ok_secs_ago,omitempty"` // seconds since the last successful store-type operation; absent if none this process + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *HealthCheckResponse) Reset() { @@ -157,12 +162,47 @@ func (x *HealthCheckResponse) GetPaymentVaultAddress() string { return "" } +func (x *HealthCheckResponse) GetWriteReady() bool { + if x != nil { + return x.WriteReady + } + return false +} + +func (x *HealthCheckResponse) GetConnectedPeers() uint32 { + if x != nil { + return x.ConnectedPeers + } + return 0 +} + +func (x *HealthCheckResponse) GetRoutingTableSize() uint32 { + if x != nil { + return x.RoutingTableSize + } + return 0 +} + +func (x *HealthCheckResponse) GetRebootstrapThreshold() uint32 { + if x != nil { + return x.RebootstrapThreshold + } + return 0 +} + +func (x *HealthCheckResponse) GetLastStoreOkSecsAgo() uint64 { + if x != nil && x.LastStoreOkSecsAgo != nil { + return *x.LastStoreOkSecsAgo + } + return 0 +} + var File_antd_v1_health_proto protoreflect.FileDescriptor const file_antd_v1_health_proto_rawDesc = "" + "\n" + "\x14antd/v1/health.proto\x12\aantd.v1\"\x14\n" + - "\x12HealthCheckRequest\"\xb4\x02\n" + + "\x12HealthCheckRequest\"\xb5\x04\n" + "\x13HealthCheckResponse\x12\x16\n" + "\x06status\x18\x01 \x01(\tR\x06status\x12\x18\n" + "\anetwork\x18\x02 \x01(\tR\anetwork\x12\x18\n" + @@ -172,7 +212,15 @@ const file_antd_v1_health_proto_rawDesc = "" + "\x0euptime_seconds\x18\x05 \x01(\x04R\ruptimeSeconds\x12!\n" + "\fbuild_commit\x18\x06 \x01(\tR\vbuildCommit\x122\n" + "\x15payment_token_address\x18\a \x01(\tR\x13paymentTokenAddress\x122\n" + - "\x15payment_vault_address\x18\b \x01(\tR\x13paymentVaultAddress2S\n" + + "\x15payment_vault_address\x18\b \x01(\tR\x13paymentVaultAddress\x12\x1f\n" + + "\vwrite_ready\x18\t \x01(\bR\n" + + "writeReady\x12'\n" + + "\x0fconnected_peers\x18\n" + + " \x01(\rR\x0econnectedPeers\x12,\n" + + "\x12routing_table_size\x18\v \x01(\rR\x10routingTableSize\x123\n" + + "\x15rebootstrap_threshold\x18\f \x01(\rR\x14rebootstrapThreshold\x127\n" + + "\x16last_store_ok_secs_ago\x18\r \x01(\x04H\x00R\x12lastStoreOkSecsAgo\x88\x01\x01B\x19\n" + + "\x17_last_store_ok_secs_ago2S\n" + "\rHealthService\x12B\n" + "\x05Check\x12\x1b.antd.v1.HealthCheckRequest\x1a\x1c.antd.v1.HealthCheckResponseBDZ8github.com/WithAutonomi/ant-sdk/antd-go/proto/antd/v1;v1\xaa\x02\aAntd.V1b\x06proto3" @@ -208,6 +256,7 @@ func file_antd_v1_health_proto_init() { if File_antd_v1_health_proto != nil { return } + file_antd_v1_health_proto_msgTypes[1].OneofWrappers = []any{} type x struct{} out := protoimpl.TypeBuilder{ File: protoimpl.DescBuilder{ diff --git a/antd-go/proto/antd/v1/upload.pb.go b/antd-go/proto/antd/v1/upload.pb.go index f702e8d0..cd1ddeef 100644 --- a/antd-go/proto/antd/v1/upload.pb.go +++ b/antd-go/proto/antd/v1/upload.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.36.11 -// protoc v4.25.3 +// protoc-gen-go v1.36.12 +// protoc v7.34.1 // source: antd/v1/upload.proto package v1 @@ -511,9 +511,21 @@ func (x *CandidateNodeEntry) GetAmount() string { return "" } +// Phase 2 of the external-signer upload. Validation happens before the +// stored upload is consumed, so a bad request leaves the paid-for upload in +// place. A storage shortfall AFTER payment is retryable against the same +// payment: the call fails with ABORTED (message prefixed "Partial upload:", +// carrying the stored/failed/total counts and a "paid attempt retained" +// hint), the daemon keeps the payment proofs plus the unstored chunks under +// the same upload_id, and repeating FinalizeUpload with that upload_id stores +// the remainder — no re-prepare, no second signature, no double payment. The +// payment fields are ignored on such a resume. Bound the retry loop: a +// persistent failure comes back as ABORTED on every call, and the retained +// attempt expires with the daemon's pending-upload TTL. type FinalizeUploadRequest struct { state protoimpl.MessageState `protogen:"open.v1"` - // The upload_id returned from a Prepare* call. + // The upload_id returned from a Prepare* call — or from a previous + // FinalizeUpload that failed with ABORTED after payment (see above). UploadId string `protobuf:"bytes,1,opt,name=upload_id,json=uploadId,proto3" json:"upload_id,omitempty"` // Wave-batch: map of quote_hash (hex) → tx_hash (hex) from the on-chain // payment. Required when the prepared upload was wave-batch and prepare @@ -530,7 +542,10 @@ type FinalizeUploadRequest struct { // `merkle_batches`, index-aligned. An empty string marks a batch the // signer never paid — paid batches store and the unpaid chunks surface // via the PARTIAL_UPLOAD error. Required (over `winner_pool_hash`) when - // the prepared upload has more than one batch. + // the prepared upload has more than one batch. Only a fully paid list + // (no empty entries) takes the resumable path: with unpaid batches the + // shortfall is NOT retryable under the same upload_id (nothing is + // retained), because a resume can never acquire proofs for unpaid chunks. WinnerPoolHashes []string `protobuf:"bytes,5,rep,name=winner_pool_hashes,json=winnerPoolHashes,proto3" json:"winner_pool_hashes,omitempty"` // If true, store the DataMap on-network via the daemon's internal wallet // and return its address. If false (default), return the raw DataMap diff --git a/antd-go/proto/antd/v1/upload_grpc.pb.go b/antd-go/proto/antd/v1/upload_grpc.pb.go index 9eb9365b..d75ae8a0 100644 --- a/antd-go/proto/antd/v1/upload_grpc.pb.go +++ b/antd-go/proto/antd/v1/upload_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 -// - protoc v4.25.3 +// - protoc v7.34.1 // source: antd/v1/upload.proto package v1 diff --git a/antd-go/range.go b/antd-go/range.go new file mode 100644 index 00000000..8165009a --- /dev/null +++ b/antd-go/range.go @@ -0,0 +1,98 @@ +package antd + +import ( + "errors" + "fmt" + "io" + "math" + "strconv" + "strings" +) + +// RangeToEnd, passed as the length to the *Range download methods, reads from +// the offset to the end of the object. +const RangeToEnd int64 = -1 + +// RangeReader is a byte-range download stream, returned by DataStreamRange and +// DataStreamPublicRange on both the REST and gRPC clients. Read it like any +// io.ReadCloser and Close it when done. The daemon fetches and decrypts only +// the chunks that overlap the range. +// +// Offset and Length describe the bytes the stream carries, after the daemon +// clamps a length past the end of the object; Size is the whole object's +// plaintext size. A stream that ends before Length bytes signals a failed +// download. +type RangeReader struct { + io.ReadCloser + Offset int64 + Length int64 + Size int64 +} + +// errNoRangeSupport reports a daemon that answered a range request with the +// whole object, which only a daemon predating byte-range support does. +var errNoRangeSupport = errors.New("antd: the daemon ignored the byte range (it predates range support); upgrade antd") + +// rangeHeader validates offset/length and renders them as an HTTP Range value. +func rangeHeader(offset, length int64) (string, error) { + if err := checkRange(offset, length); err != nil { + return "", err + } + // offset+length-1 would overflow int64 for a huge length; such a range + // reads to the end anyway. + if length == RangeToEnd || length > math.MaxInt64-offset { + return fmt.Sprintf("bytes=%d-", offset), nil + } + return fmt.Sprintf("bytes=%d-%d", offset, offset+length-1), nil +} + +// checkRange rejects a negative offset and a length that is neither positive +// nor RangeToEnd, before any request is sent. +func checkRange(offset, length int64) error { + if offset < 0 { + return fmt.Errorf("antd: range offset must not be negative, got %d", offset) + } + if length <= 0 && length != RangeToEnd { + return fmt.Errorf("antd: range length must be positive or RangeToEnd, got %d", length) + } + return nil +} + +// parseContentRange parses a satisfied range, "bytes first-last/size", into +// a RangeReader's Offset, Length and Size. +func parseContentRange(value string) (*RangeReader, error) { + bad := fmt.Errorf("antd: malformed Content-Range %q", value) + spec, ok := strings.CutPrefix(strings.TrimSpace(value), "bytes ") + if !ok { + return nil, bad + } + span, size, ok := strings.Cut(spec, "/") + if !ok { + return nil, bad + } + first, last, ok := strings.Cut(span, "-") + if !ok { + return nil, bad + } + f, err1 := strconv.ParseInt(first, 10, 64) + l, err2 := strconv.ParseInt(last, 10, 64) + s, err3 := strconv.ParseInt(size, 10, 64) + if err1 != nil || err2 != nil || err3 != nil || f < 0 || l < f || l >= s { + return nil, bad + } + return &RangeReader{Offset: f, Length: l - f + 1, Size: s}, nil +} + +// unsatisfiedSize parses the size from a 416's "bytes */size" Content-Range, +// or returns -1 when it is absent or malformed. +func unsatisfiedSize(value string) int64 { + size, ok := strings.CutPrefix(strings.TrimSpace(value), "bytes */") + if !ok { + return -1 + } + s, err := strconv.ParseInt(size, 10, 64) + if err != nil || s < 0 { + return -1 + } + return s +} diff --git a/antd-go/range_test.go b/antd-go/range_test.go new file mode 100644 index 00000000..a9ca2218 --- /dev/null +++ b/antd-go/range_test.go @@ -0,0 +1,206 @@ +package antd + +import ( + "context" + "errors" + "fmt" + "io" + "math" + "net/http" + "net/http/httptest" + "testing" + + pb "github.com/WithAutonomi/ant-sdk/antd-go/proto/antd/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +func TestRangeHeader(t *testing.T) { + for _, tc := range []struct { + offset, length int64 + want string + }{ + {0, 10, "bytes=0-9"}, + {5, 1, "bytes=5-5"}, + {7, RangeToEnd, "bytes=7-"}, + {1, math.MaxInt64, "bytes=1-"}, + } { + got, err := rangeHeader(tc.offset, tc.length) + if err != nil || got != tc.want { + t.Errorf("rangeHeader(%d, %d) = %q, %v; want %q", tc.offset, tc.length, got, err, tc.want) + } + } + for _, tc := range [][2]int64{{-1, 10}, {0, 0}, {0, -2}} { + if _, err := rangeHeader(tc[0], tc[1]); err == nil { + t.Errorf("rangeHeader(%d, %d) accepted an invalid range", tc[0], tc[1]) + } + } +} + +func TestParseContentRange(t *testing.T) { + r, err := parseContentRange("bytes 100-199/1000") + if err != nil || r.Offset != 100 || r.Length != 100 || r.Size != 1000 { + t.Fatalf("got %+v, %v", r, err) + } + for _, bad := range []string{"", "bytes */1000", "bytes 5-4/10", "bytes 0-10/10", "items 0-1/2", "bytes 0-1"} { + if _, err := parseContentRange(bad); err == nil { + t.Errorf("parseContentRange(%q) accepted a malformed value", bad) + } + } + if unsatisfiedSize("bytes */1234") != 1234 || unsatisfiedSize("") != -1 { + t.Error("unsatisfiedSize") + } +} + +// rangeServer serves payload on both stream routes the way the daemon does: +// 206 + Content-Range for a satisfiable Range, 416 + "bytes */size" past the +// end. legacy makes it ignore Range and answer 200, like an old daemon. +func rangeServer(t *testing.T, payload string, legacy bool) (*Client, *string) { + t.Helper() + var gotRange string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotRange = r.Header.Get("Range") + if legacy { + _, _ = io.WriteString(w, payload) + return + } + var start, last int64 + size := int64(len(payload)) + if n, _ := fmt.Sscanf(gotRange, "bytes=%d-%d", &start, &last); n == 1 { + last = size - 1 + } + if start >= size { + w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", size)) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusRequestedRangeNotSatisfiable) + _, _ = io.WriteString(w, `{"error":"Range not satisfiable","code":"RANGE_NOT_SATISFIABLE"}`) + return + } + last = min(last, size-1) + w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, last, size)) + w.WriteHeader(http.StatusPartialContent) + _, _ = io.WriteString(w, payload[start:last+1]) + })) + t.Cleanup(srv.Close) + return NewClient(srv.URL), &gotRange +} + +func readRange(t *testing.T, r *RangeReader, err error) string { + t.Helper() + if err != nil { + t.Fatal(err) + } + defer r.Close() + data, err := io.ReadAll(r) + if err != nil { + t.Fatal(err) + } + return string(data) +} + +func TestDataStreamRange(t *testing.T) { + c, gotRange := rangeServer(t, "0123456789", false) + r, err := c.DataStreamRange(context.Background(), "dm", 2, 3) + if got := readRange(t, r, err); got != "234" || *gotRange != "bytes=2-4" { + t.Fatalf("got %q with Range %q", got, *gotRange) + } + if r.Offset != 2 || r.Length != 3 || r.Size != 10 { + t.Fatalf("unexpected range metadata %+v", r) + } + + // Clamped past the end, and to the end. + r, err = c.DataStreamPublicRange(context.Background(), "addr", 8, 100) + if got := readRange(t, r, err); got != "89" || r.Length != 2 { + t.Fatalf("clamped: got %q, %+v", got, r) + } + r, err = c.DataStreamRange(context.Background(), "dm", 7, RangeToEnd) + if got := readRange(t, r, err); got != "789" || *gotRange != "bytes=7-" { + t.Fatalf("to end: got %q with Range %q", got, *gotRange) + } +} + +func TestDataStreamRangeNotSatisfiable(t *testing.T) { + c, _ := rangeServer(t, "0123456789", false) + _, err := c.DataStreamRange(context.Background(), "dm", 10, 1) + var rerr *RangeNotSatisfiableError + if !errors.As(err, &rerr) || rerr.Size != 10 || rerr.StatusCode != 416 { + t.Fatalf("want *RangeNotSatisfiableError with Size 10, got %#v", err) + } +} + +func TestDataStreamRangeRejectsIgnoredRange(t *testing.T) { + c, _ := rangeServer(t, "0123456789", true) + if _, err := c.DataStreamRange(context.Background(), "dm", 2, 3); !errors.Is(err, errNoRangeSupport) { + t.Fatalf("a 200 must not pass for a range: got %v", err) + } +} + +func TestDataStreamRangeValidatesBeforeSending(t *testing.T) { + c, gotRange := rangeServer(t, "0123456789", false) + if _, err := c.DataStreamRange(context.Background(), "dm", 0, 0); err == nil || *gotRange != "" { + t.Fatalf("zero length must fail client-side: err %v, sent Range %q", err, *gotRange) + } +} + +// serveMockRange answers a ranged mock stream like the daemon: x-content-range +// metadata and only the requested bytes, or OUT_OF_RANGE past the end. It +// reports whether the request was ranged. The data map / address "legacy" +// ignores the range, like an old daemon. +func serveMockRange(payload string, key string, offset, length *uint64, srv grpc.ServerStreamingServer[pb.DataChunk]) (bool, error) { + if (offset == nil && length == nil) || key == "legacy" { + return false, nil + } + size := uint64(len(payload)) + start := uint64(0) + if offset != nil { + start = *offset + } + if start >= size { + return true, status.Errorf(codes.OutOfRange, "Range not satisfiable: the object is %d bytes", size) + } + end := size + if length != nil { + end = min(start+*length, size) + } + _ = srv.SetHeader(metadata.Pairs( + "x-content-length", fmt.Sprint(end-start), + "x-content-range", fmt.Sprintf("bytes %d-%d/%d", start, end-1, size), + )) + // One byte per frame, so the reader's frame buffering is exercised. + for i := start; i < end; i++ { + if err := srv.Send(&pb.DataChunk{Kind: &pb.DataChunk_Data{Data: []byte{payload[i]}}}); err != nil { + return true, err + } + } + return true, nil +} + +func TestGrpcDataStreamRange(t *testing.T) { + c := startMockServer(t) + r, err := c.DataStreamRange(context.Background(), "dm123", 1, 3) + if got := readRange(t, r, err); got != "ecr" || r.Offset != 1 || r.Length != 3 || r.Size != 6 { + t.Fatalf("got %q, %+v", got, r) + } + r, err = c.DataStreamPublicRange(context.Background(), "abc123", 3, RangeToEnd) + if got := readRange(t, r, err); got != "lo" || r.Size != 5 { + t.Fatalf("public to end: got %q, %+v", got, r) + } +} + +func TestGrpcDataStreamRangeOutOfRange(t *testing.T) { + c := startMockServer(t) + _, err := c.DataStreamRange(context.Background(), "dm123", 6, RangeToEnd) + var rerr *RangeNotSatisfiableError + if !errors.As(err, &rerr) || rerr.StatusCode != 416 { + t.Fatalf("want *RangeNotSatisfiableError, got %#v", err) + } +} + +func TestGrpcDataStreamRangeRejectsIgnoredRange(t *testing.T) { + c := startMockServer(t) + if _, err := c.DataStreamRange(context.Background(), "legacy", 1, 3); !errors.Is(err, errNoRangeSupport) { + t.Fatalf("a stream without x-content-range must not pass for a range: got %v", err) + } +} diff --git a/antd-kotlin/lib/src/main/proto/antd/v1/data.proto b/antd-kotlin/lib/src/main/proto/antd/v1/data.proto index 8b23f8fb..a30aad17 100644 --- a/antd-kotlin/lib/src/main/proto/antd/v1/data.proto +++ b/antd-kotlin/lib/src/main/proto/antd/v1/data.proto @@ -57,6 +57,16 @@ message StreamPublicDataRequest { // progress bar. Defaults to false — old clients omitting it receive a // pure data-frame stream, byte-identical to the pre-progress behaviour. bool include_progress = 2; + // Byte-range download. Setting either field streams only the plaintext bytes + // [offset, offset + length) instead of the whole object, fetching just the + // chunks that overlap them. `offset` defaults to 0 and an absent `length` + // reads to the end; a `length` past the end is clamped. `length` 0 is + // INVALID_ARGUMENT, and an `offset` at or past the end is OUT_OF_RANGE. A + // ranged stream's `x-content-length` metadata is the range's byte count, and + // it adds `x-content-range` (`bytes first-last/size`, `last` inclusive). + // Cannot be combined with include_progress (INVALID_ARGUMENT). + optional uint64 offset = 3; + optional uint64 length = 4; } // A single frame of a streaming download. Exactly one of `data` (a decrypted @@ -98,6 +108,9 @@ message StreamDataRequest { string data_map = 1; // hex // See StreamPublicDataRequest.include_progress. Defaults to false. bool include_progress = 2; + // Byte range: see StreamPublicDataRequest.offset / length. + optional uint64 offset = 3; + optional uint64 length = 4; } message GetDataResponse { diff --git a/antd-rust/src/generated/antd.v1.rs b/antd-rust/src/generated/antd.v1.rs index 5d2ef803..3f69dd65 100644 --- a/antd-rust/src/generated/antd.v1.rs +++ b/antd-rust/src/generated/antd.v1.rs @@ -452,6 +452,18 @@ pub struct StreamPublicDataRequest { /// pure data-frame stream, byte-identical to the pre-progress behaviour. #[prost(bool, tag = "2")] pub include_progress: bool, + /// Byte-range download. Setting either field streams only the plaintext bytes + /// [offset, offset + length) instead of the whole object, fetching just the + /// chunks that overlap them. `offset` defaults to 0 and an absent `length` + /// reads to the end; a `length` past the end is clamped. `length` 0 is + /// INVALID_ARGUMENT, and an `offset` at or past the end is OUT_OF_RANGE. A + /// ranged stream's `x-content-length` metadata is the range's byte count, and + /// it adds `x-content-range` (`bytes first-last/size`, `last` inclusive). + /// Cannot be combined with include_progress (INVALID_ARGUMENT). + #[prost(uint64, optional, tag = "3")] + pub offset: ::core::option::Option, + #[prost(uint64, optional, tag = "4")] + pub length: ::core::option::Option, } /// A single frame of a streaming download. Exactly one of `data` (a decrypted /// plaintext chunk) or `progress` (a fetch-progress update) is set per frame. @@ -507,6 +519,11 @@ pub struct StreamDataRequest { /// See StreamPublicDataRequest.include_progress. Defaults to false. #[prost(bool, tag = "2")] pub include_progress: bool, + /// Byte range: see StreamPublicDataRequest.offset / length. + #[prost(uint64, optional, tag = "3")] + pub offset: ::core::option::Option, + #[prost(uint64, optional, tag = "4")] + pub length: ::core::option::Option, } #[derive(Clone, PartialEq, ::prost::Message)] pub struct GetDataResponse { diff --git a/antd-rust/src/grpc_client.rs b/antd-rust/src/grpc_client.rs index 34b04ace..488ac4eb 100644 --- a/antd-rust/src/grpc_client.rs +++ b/antd-rust/src/grpc_client.rs @@ -266,6 +266,7 @@ impl GrpcClient { .stream(proto::antd::v1::StreamDataRequest { data_map: data_map.to_string(), include_progress: false, + ..Default::default() }) .await? .into_inner(); @@ -289,6 +290,7 @@ impl GrpcClient { .stream_public(proto::antd::v1::StreamPublicDataRequest { address: address.to_string(), include_progress: false, + ..Default::default() }) .await? .into_inner(); @@ -316,6 +318,7 @@ impl GrpcClient { .stream(proto::antd::v1::StreamDataRequest { data_map: data_map.to_string(), include_progress: true, + ..Default::default() }) .await?; @@ -340,6 +343,7 @@ impl GrpcClient { .stream_public(proto::antd::v1::StreamPublicDataRequest { address: address.to_string(), include_progress: true, + ..Default::default() }) .await?; diff --git a/antd-swift/Sources/AntdSdk/Proto/antd/v1/data.pb.swift b/antd-swift/Sources/AntdSdk/Proto/antd/v1/data.pb.swift index e7ab09e3..02b9cd4c 100644 --- a/antd-swift/Sources/AntdSdk/Proto/antd/v1/data.pb.swift +++ b/antd-swift/Sources/AntdSdk/Proto/antd/v1/data.pb.swift @@ -111,9 +111,38 @@ public struct Antd_V1_StreamPublicDataRequest: Sendable { /// pure data-frame stream, byte-identical to the pre-progress behaviour. public var includeProgress: Bool = false + /// Byte-range download. Setting either field streams only the plaintext bytes + /// [offset, offset + length) instead of the whole object, fetching just the + /// chunks that overlap them. `offset` defaults to 0 and an absent `length` + /// reads to the end; a `length` past the end is clamped. `length` 0 is + /// INVALID_ARGUMENT, and an `offset` at or past the end is OUT_OF_RANGE. A + /// ranged stream's `x-content-length` metadata is the range's byte count, and + /// it adds `x-content-range` (`bytes first-last/size`, `last` inclusive). + /// Cannot be combined with include_progress (INVALID_ARGUMENT). + public var offset: UInt64 { + get {_offset ?? 0} + set {_offset = newValue} + } + /// Returns true if `offset` has been explicitly set. + public var hasOffset: Bool {self._offset != nil} + /// Clears the value of `offset`. Subsequent reads from it will return its default value. + public mutating func clearOffset() {self._offset = nil} + + public var length: UInt64 { + get {_length ?? 0} + set {_length = newValue} + } + /// Returns true if `length` has been explicitly set. + public var hasLength: Bool {self._length != nil} + /// Clears the value of `length`. Subsequent reads from it will return its default value. + public mutating func clearLength() {self._length = nil} + public var unknownFields = SwiftProtobuf.UnknownStorage() public init() {} + + fileprivate var _offset: UInt64? = nil + fileprivate var _length: UInt64? = nil } /// A single frame of a streaming download. Exactly one of `data` (a decrypted @@ -206,9 +235,31 @@ public struct Antd_V1_StreamDataRequest: Sendable { /// See StreamPublicDataRequest.include_progress. Defaults to false. public var includeProgress: Bool = false + /// Byte range: see StreamPublicDataRequest.offset / length. + public var offset: UInt64 { + get {_offset ?? 0} + set {_offset = newValue} + } + /// Returns true if `offset` has been explicitly set. + public var hasOffset: Bool {self._offset != nil} + /// Clears the value of `offset`. Subsequent reads from it will return its default value. + public mutating func clearOffset() {self._offset = nil} + + public var length: UInt64 { + get {_length ?? 0} + set {_length = newValue} + } + /// Returns true if `length` has been explicitly set. + public var hasLength: Bool {self._length != nil} + /// Clears the value of `length`. Subsequent reads from it will return its default value. + public mutating func clearLength() {self._length = nil} + public var unknownFields = SwiftProtobuf.UnknownStorage() public init() {} + + fileprivate var _offset: UInt64? = nil + fileprivate var _length: UInt64? = nil } public struct Antd_V1_GetDataResponse: Sendable { @@ -436,7 +487,7 @@ extension Antd_V1_PutPublicDataResponse: SwiftProtobuf.Message, SwiftProtobuf._M extension Antd_V1_StreamPublicDataRequest: SwiftProtobuf.Message, SwiftProtobuf._MessageImplementationBase, SwiftProtobuf._ProtoNameProviding { public static let protoMessageName: String = _protobuf_package + ".StreamPublicDataRequest" - public static let _protobuf_nameMap = SwiftProtobuf._NameMap(bytecode: "\0\u{1}address\0\u{3}include_progress\0") + public static let _protobuf_nameMap = SwiftProtobuf._NameMap(bytecode: "\0\u{1}address\0\u{3}include_progress\0\u{1}offset\0\u{1}length\0") public mutating func decodeMessage(decoder: inout D) throws { while let fieldNumber = try decoder.nextFieldNumber() { @@ -446,24 +497,38 @@ extension Antd_V1_StreamPublicDataRequest: SwiftProtobuf.Message, SwiftProtobuf. switch fieldNumber { case 1: try { try decoder.decodeSingularStringField(value: &self.address) }() case 2: try { try decoder.decodeSingularBoolField(value: &self.includeProgress) }() + case 3: try { try decoder.decodeSingularUInt64Field(value: &self._offset) }() + case 4: try { try decoder.decodeSingularUInt64Field(value: &self._length) }() default: break } } } public func traverse(visitor: inout V) throws { + // The use of inline closures is to circumvent an issue where the compiler + // allocates stack space for every if/case branch local when no optimizations + // are enabled. https://github.com/apple/swift-protobuf/issues/1034 and + // https://github.com/apple/swift-protobuf/issues/1182 if !self.address.isEmpty { try visitor.visitSingularStringField(value: self.address, fieldNumber: 1) } if self.includeProgress != false { try visitor.visitSingularBoolField(value: self.includeProgress, fieldNumber: 2) } + try { if let v = self._offset { + try visitor.visitSingularUInt64Field(value: v, fieldNumber: 3) + } }() + try { if let v = self._length { + try visitor.visitSingularUInt64Field(value: v, fieldNumber: 4) + } }() try unknownFields.traverse(visitor: &visitor) } public static func ==(lhs: Antd_V1_StreamPublicDataRequest, rhs: Antd_V1_StreamPublicDataRequest) -> Bool { if lhs.address != rhs.address {return false} if lhs.includeProgress != rhs.includeProgress {return false} + if lhs._offset != rhs._offset {return false} + if lhs._length != rhs._length {return false} if lhs.unknownFields != rhs.unknownFields {return false} return true } @@ -603,7 +668,7 @@ extension Antd_V1_GetDataRequest: SwiftProtobuf.Message, SwiftProtobuf._MessageI extension Antd_V1_StreamDataRequest: SwiftProtobuf.Message, SwiftProtobuf._MessageImplementationBase, SwiftProtobuf._ProtoNameProviding { public static let protoMessageName: String = _protobuf_package + ".StreamDataRequest" - public static let _protobuf_nameMap = SwiftProtobuf._NameMap(bytecode: "\0\u{3}data_map\0\u{3}include_progress\0") + public static let _protobuf_nameMap = SwiftProtobuf._NameMap(bytecode: "\0\u{3}data_map\0\u{3}include_progress\0\u{1}offset\0\u{1}length\0") public mutating func decodeMessage(decoder: inout D) throws { while let fieldNumber = try decoder.nextFieldNumber() { @@ -613,24 +678,38 @@ extension Antd_V1_StreamDataRequest: SwiftProtobuf.Message, SwiftProtobuf._Messa switch fieldNumber { case 1: try { try decoder.decodeSingularStringField(value: &self.dataMap) }() case 2: try { try decoder.decodeSingularBoolField(value: &self.includeProgress) }() + case 3: try { try decoder.decodeSingularUInt64Field(value: &self._offset) }() + case 4: try { try decoder.decodeSingularUInt64Field(value: &self._length) }() default: break } } } public func traverse(visitor: inout V) throws { + // The use of inline closures is to circumvent an issue where the compiler + // allocates stack space for every if/case branch local when no optimizations + // are enabled. https://github.com/apple/swift-protobuf/issues/1034 and + // https://github.com/apple/swift-protobuf/issues/1182 if !self.dataMap.isEmpty { try visitor.visitSingularStringField(value: self.dataMap, fieldNumber: 1) } if self.includeProgress != false { try visitor.visitSingularBoolField(value: self.includeProgress, fieldNumber: 2) } + try { if let v = self._offset { + try visitor.visitSingularUInt64Field(value: v, fieldNumber: 3) + } }() + try { if let v = self._length { + try visitor.visitSingularUInt64Field(value: v, fieldNumber: 4) + } }() try unknownFields.traverse(visitor: &visitor) } public static func ==(lhs: Antd_V1_StreamDataRequest, rhs: Antd_V1_StreamDataRequest) -> Bool { if lhs.dataMap != rhs.dataMap {return false} if lhs.includeProgress != rhs.includeProgress {return false} + if lhs._offset != rhs._offset {return false} + if lhs._length != rhs._length {return false} if lhs.unknownFields != rhs.unknownFields {return false} return true } diff --git a/antd/openapi.yaml b/antd/openapi.yaml index afd12572..5efffd9e 100644 --- a/antd/openapi.yaml +++ b/antd/openapi.yaml @@ -65,7 +65,10 @@ paths: tags: [data] operationId: dataStreamPublic summary: Stream public data - description: Stream public data as a chunked response. + description: >- + Stream public data as a chunked response. With a `Range` header, only + that byte range is streamed and only the chunks overlapping it are + fetched. parameters: - name: addr in: path @@ -73,11 +76,71 @@ paths: schema: type: string description: Hex address of the data + - $ref: "#/components/parameters/Range" responses: "200": - description: Chunked data stream + description: >- + The whole object as `application/octet-stream`. `Content-Length` + is the plaintext size, so a body that ends short signals a failed + download. Carries `Accept-Ranges: bytes`. + "206": + description: >- + The requested byte range (`Range` request). `Content-Range` is + `bytes first-last/size` and `Content-Length` is the range's byte + count; a body that ends short signals a failed download. + headers: + Content-Range: + schema: + type: string + description: "`bytes first-last/size` (`last` inclusive)" + "400": + $ref: "#/components/responses/BadRequest" "404": $ref: "#/components/responses/NotFound" + "416": + $ref: "#/components/responses/RangeNotSatisfiable" + + /v1/data/stream: + post: + tags: [data] + operationId: dataStream + summary: Stream private data + description: >- + Stream private data from a caller-held DataMap instead of buffering it + into a base64 JSON body. POST because the hex DataMap can be many KB. + With a `Range` header, only that byte range is streamed and only the + chunks overlapping it are fetched (RFC 9110 defines `Range` only for + GET; antd honours it on this route too). + parameters: + - $ref: "#/components/parameters/Range" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/DataGetRequest" + responses: + "200": + description: >- + The whole object as `application/octet-stream`. `Content-Length` + is the plaintext size, so a body that ends short signals a failed + download. Carries `Accept-Ranges: bytes`. + "206": + description: >- + The requested byte range (`Range` request). `Content-Range` is + `bytes first-last/size` and `Content-Length` is the range's byte + count; a body that ends short signals a failed download. + headers: + Content-Range: + schema: + type: string + description: "`bytes first-last/size` (`last` inclusive)" + "400": + $ref: "#/components/responses/BadRequest" + "404": + $ref: "#/components/responses/NotFound" + "416": + $ref: "#/components/responses/RangeNotSatisfiable" /v1/data/public: post: @@ -653,6 +716,21 @@ paths: $ref: "#/components/responses/PartialUpload" components: + parameters: + Range: + name: Range + in: header + required: false + schema: + type: string + example: bytes=0-1048575 + description: >- + One byte range: `bytes=first-last`, `bytes=first-` (to the end) or + `bytes=-n` (the last n bytes). A `last` past the end is clamped. A + range starting at or past the end is 416; a malformed range, more + than one range, or a Range combined with `Accept: + application/x-ndjson` is 400. + schemas: # -- Health -- HealthResponse: @@ -1464,6 +1542,20 @@ components: application/json: schema: $ref: "#/components/schemas/ErrorResponse" + RangeNotSatisfiable: + description: >- + The range starts at or past the end of the object (416). Code + RANGE_NOT_SATISFIABLE; `Content-Range: bytes */size` gives the + object's size. + headers: + Content-Range: + schema: + type: string + description: "`bytes */size`" + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" NetworkError: description: Network unreachable (502) content: diff --git a/antd/proto/antd/v1/data.proto b/antd/proto/antd/v1/data.proto index 8b23f8fb..a30aad17 100644 --- a/antd/proto/antd/v1/data.proto +++ b/antd/proto/antd/v1/data.proto @@ -57,6 +57,16 @@ message StreamPublicDataRequest { // progress bar. Defaults to false — old clients omitting it receive a // pure data-frame stream, byte-identical to the pre-progress behaviour. bool include_progress = 2; + // Byte-range download. Setting either field streams only the plaintext bytes + // [offset, offset + length) instead of the whole object, fetching just the + // chunks that overlap them. `offset` defaults to 0 and an absent `length` + // reads to the end; a `length` past the end is clamped. `length` 0 is + // INVALID_ARGUMENT, and an `offset` at or past the end is OUT_OF_RANGE. A + // ranged stream's `x-content-length` metadata is the range's byte count, and + // it adds `x-content-range` (`bytes first-last/size`, `last` inclusive). + // Cannot be combined with include_progress (INVALID_ARGUMENT). + optional uint64 offset = 3; + optional uint64 length = 4; } // A single frame of a streaming download. Exactly one of `data` (a decrypted @@ -98,6 +108,9 @@ message StreamDataRequest { string data_map = 1; // hex // See StreamPublicDataRequest.include_progress. Defaults to false. bool include_progress = 2; + // Byte range: see StreamPublicDataRequest.offset / length. + optional uint64 offset = 3; + optional uint64 length = 4; } message GetDataResponse { diff --git a/antd/src/error.rs b/antd/src/error.rs index af9eaf6a..06527631 100644 --- a/antd/src/error.rs +++ b/antd/src/error.rs @@ -34,6 +34,12 @@ pub enum AntdError { #[error("Internal error: {0}")] Internal(String), + /// A byte-range download (V2-1425) that starts at or past the end of the + /// object. `size` is the object's plaintext size, reported back so the + /// caller can re-issue a satisfiable range. + #[error("Range not satisfiable: the object is {size} bytes")] + RangeNotSatisfiable { size: u64 }, + /// Upload partially succeeded: some chunks stored, some failed quorum /// after all retries. The payment was made and the stored chunks persist. /// @@ -87,6 +93,7 @@ impl AntdError { AntdError::ServiceUnavailable(_) => "SERVICE_UNAVAILABLE", AntdError::NotImplemented(_) => "NOT_IMPLEMENTED", AntdError::Internal(_) => "INTERNAL_ERROR", + AntdError::RangeNotSatisfiable { .. } => "RANGE_NOT_SATISFIABLE", AntdError::PartialUpload { .. } => "PARTIAL_UPLOAD", } } @@ -165,6 +172,7 @@ impl IntoResponse for AntdError { AntdError::ServiceUnavailable(_) => StatusCode::SERVICE_UNAVAILABLE, AntdError::NotImplemented(_) => StatusCode::NOT_IMPLEMENTED, AntdError::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR, + AntdError::RangeNotSatisfiable { .. } => StatusCode::RANGE_NOT_SATISFIABLE, // The upstream network failed to store part of the file; the // request itself was valid, so this is a gateway-side failure. AntdError::PartialUpload { .. } => StatusCode::BAD_GATEWAY, @@ -188,12 +196,21 @@ impl IntoResponse for AntdError { retryable, }) .unwrap_or_else(|_| r#"{"error":"internal error","code":"INTERNAL_ERROR"}"#.to_string()); - ( + let mut response = ( status, [(axum::http::header::CONTENT_TYPE, "application/json")], body, ) - .into_response() + .into_response(); + // RFC 9110 §15.5.17: a 416 carries the current length as `bytes */size`. + if let AntdError::RangeNotSatisfiable { size } = self { + response.headers_mut().insert( + axum::http::header::CONTENT_RANGE, + axum::http::HeaderValue::from_str(&format!("bytes */{size}")) + .expect("decimal digits are a valid header value"), + ); + } + response } } @@ -210,6 +227,7 @@ impl From for tonic::Status { AntdError::ServiceUnavailable(msg) => tonic::Status::unavailable(msg), AntdError::NotImplemented(msg) => tonic::Status::unimplemented(msg), AntdError::Internal(msg) => tonic::Status::internal(msg), + e @ AntdError::RangeNotSatisfiable { .. } => tonic::Status::out_of_range(e.to_string()), // ABORTED: the operation stopped partway and the retry lives at // the application level — a repeat FinalizeUpload with the same // upload_id when the paid attempt was retained (`retryable`), or @@ -243,6 +261,25 @@ mod tests { assert_eq!(resp.status(), StatusCode::NOT_FOUND); } + #[test] + fn range_not_satisfiable_is_http_416_with_content_range() { + let resp = AntdError::RangeNotSatisfiable { size: 1234 }.into_response(); + assert_eq!(resp.status(), StatusCode::RANGE_NOT_SATISFIABLE); + assert_eq!( + resp.headers() + .get(axum::http::header::CONTENT_RANGE) + .unwrap(), + "bytes */1234" + ); + } + + #[test] + fn range_not_satisfiable_is_grpc_out_of_range() { + let status = tonic::Status::from(AntdError::RangeNotSatisfiable { size: 7 }); + assert_eq!(status.code(), tonic::Code::OutOfRange); + assert!(status.message().contains("7 bytes")); + } + #[test] fn not_found_is_grpc_not_found() { let status = tonic::Status::from(AntdError::NotFound("gone".into())); diff --git a/antd/src/grpc/service.rs b/antd/src/grpc/service.rs index 504254a2..1f0bbae3 100644 --- a/antd/src/grpc/service.rs +++ b/antd/src/grpc/service.rs @@ -376,6 +376,101 @@ async fn data_chunk_stream_response( Ok(response) } +/// gRPC frame size for a ranged stream. Range windows can be ~32 MiB, which is +/// over the 4 MiB default receive limit of most gRPC clients, so each window is +/// split into frames of this size. +const RANGE_FRAME_BYTES: usize = 1024 * 1024; + +/// Build the gRPC server-streaming response for a byte-range download +/// (V2-1425): the `offset` / `length` fields of `StreamDataRequest` / +/// `StreamPublicDataRequest`. Like [`data_chunk_stream_response`] it resolves a +/// shrunk map before the stream opens and ends the stream with a terminal +/// `Status` on failure. `x-content-length` is the range's byte count, and +/// `x-content-range` (`bytes first-last/size`) carries the object size. +#[allow(clippy::result_large_err)] +async fn data_chunk_range_response( + client: Arc, + data_map: ant_core::data::DataMap, + offset: Option, + length: Option, +) -> Result>>, Status> +{ + let data_map = crate::datamap::resolve_root_data_map(&client, data_map) + .await + .map_err(|e| Status::from(AntdError::from_core(e)))?; + let size = data_map.original_file_size() as u64; + let range = crate::range::range_from_offset(offset.unwrap_or(0), length, size).map_err( + |e| match e { + crate::range::RangeError::Invalid(msg) => Status::invalid_argument(msg), + crate::range::RangeError::Unsatisfiable => { + Status::from(AntdError::RangeNotSatisfiable { size }) + } + }, + )?; + + let mut byte_rx = crate::range::spawn_range_stream(client, data_map, range); + let (out_tx, out_rx) = tokio::sync::mpsc::channel::>(16); + tokio::spawn(async move { + while let Some(item) = byte_rx.recv().await { + match item { + Ok(mut bytes) => { + while !bytes.is_empty() { + let frame = bytes.split_to(bytes.len().min(RANGE_FRAME_BYTES)); + let chunk = pb::DataChunk { + kind: Some(pb::data_chunk::Kind::Data(frame.to_vec())), + }; + if out_tx.send(Ok(chunk)).await.is_err() { + return; + } + } + } + Err(e) => { + let _ = out_tx + .send(Err(Status::from(AntdError::from_core(e)))) + .await; + return; + } + } + } + }); + + let mut response = Response::new(tokio_stream::wrappers::ReceiverStream::new(out_rx)); + let metadata = response.metadata_mut(); + metadata.insert( + "x-content-length", + range + .len() + .to_string() + .parse() + .expect("decimal digits are a valid ascii metadata value"), + ); + metadata.insert( + "x-content-range", + range + .content_range(size) + .parse() + .expect("a byte-range string is a valid ascii metadata value"), + ); + Ok(response) +} + +/// Whether a stream request asked for a byte range, rejecting a range combined +/// with `include_progress` (ranged streams carry data frames only). +#[allow(clippy::result_large_err)] +fn wants_range( + offset: Option, + length: Option, + include_progress: bool, +) -> Result { + let ranged = offset.is_some() || length.is_some(); + if ranged && include_progress { + return Err(Status::invalid_argument( + "offset/length cannot be combined with include_progress", + )); + } + Ok(ranged) +} + #[tonic::async_trait] impl pb::data_service_server::DataService for DataServiceImpl { async fn get_public( @@ -487,6 +582,9 @@ impl pb::data_service_server::DataService for DataServiceImpl { .map_err(|e| Status::invalid_argument(format!("invalid data map: {e}")))?; let client = self.state.client.clone(); + if wants_range(req.offset, req.length, include_progress)? { + return data_chunk_range_response(client, data_map, req.offset, req.length).await; + } data_chunk_stream_response(client, data_map, include_progress).await } @@ -497,6 +595,7 @@ impl pb::data_service_server::DataService for DataServiceImpl { ) -> Result, Status> { let req = request.into_inner(); let include_progress = req.include_progress; + let ranged = wants_range(req.offset, req.length, include_progress)?; let addr = req.address; if addr.len() != 64 { return Err(Status::invalid_argument( @@ -519,6 +618,9 @@ impl pb::data_service_server::DataService for DataServiceImpl { .map_err(AntdError::from_core) .map_err(tonic::Status::from)?; + if ranged { + return data_chunk_range_response(client, data_map, req.offset, req.length).await; + } data_chunk_stream_response(client, data_map, include_progress).await } diff --git a/antd/src/main.rs b/antd/src/main.rs index a2b52d8a..825072fa 100644 --- a/antd/src/main.rs +++ b/antd/src/main.rs @@ -23,6 +23,7 @@ mod evm_defaults; mod grpc; mod peers; mod port_file; +mod range; mod rest; mod signed_quotes; mod state; diff --git a/antd/src/range.rs b/antd/src/range.rs new file mode 100644 index 00000000..776d9730 --- /dev/null +++ b/antd/src/range.rs @@ -0,0 +1,357 @@ +//! Byte-range downloads (V2-1425), shared by the REST and gRPC streaming +//! handlers. +//! +//! ant-core's `Client::data_download_range` fetches and decrypts only the +//! chunks overlapping a range (their neighbours' hashes are already in the +//! DataMap, so no neighbour chunks are fetched), but it returns one buffered +//! `Bytes`. An open-ended range such as `bytes=0-` on a multi-GB object would +//! then buffer the whole object. So a range is read in windows aligned to +//! chunk boundaries: each window is one `data_download_range` call over at +//! most [`RANGE_WINDOW_CHUNKS`] chunks, so every chunk is fetched exactly once +//! and memory stays bounded however large the range is. + +use std::sync::Arc; + +use ant_core::data::{Client, DataMap, Error}; +use bytes::Bytes; +use tokio::sync::mpsc; + +/// Chunks fetched and decrypted per window. Chunks are at most ~4 MiB, so a +/// window holds at most ~32 MiB of plaintext. The producer reads the next +/// window while the previous one drains, so a ranged stream holds two or three +/// windows in memory at once. +pub const RANGE_WINDOW_CHUNKS: usize = 8; + +/// A satisfiable, non-empty, half-open plaintext byte range `[start, end)`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ByteRange { + pub start: u64, + pub end: u64, +} + +impl ByteRange { + /// Number of bytes in the range (never 0). + pub fn len(&self) -> u64 { + self.end - self.start + } + + /// The `Content-Range` value for this range of an object of `size` bytes: + /// `bytes first-last/size`, with an inclusive `last`. + pub fn content_range(&self, size: u64) -> String { + format!("bytes {}-{}/{}", self.start, self.end - 1, size) + } +} + +/// Why a requested range could not be served. +#[derive(Debug, PartialEq, Eq)] +pub enum RangeError { + /// Malformed, multi-range, or a zero/negative length: a caller bug. + Invalid(String), + /// Well-formed, but starts at or past the end of the object. + Unsatisfiable, +} + +/// Parse an HTTP `Range` header value against an object of `size` bytes. +/// +/// One `bytes` range is supported, in any of the RFC 9110 forms: `first-last`, +/// `first-` (to the end) and `-suffix` (the last `suffix` bytes). A `last` or +/// `suffix` past the end is clamped. Multiple ranges are rejected as +/// [`RangeError::Invalid`] rather than silently answered with the whole +/// object, since that could be many GB the caller didn't ask for. +pub fn parse_range_header(value: &str, size: u64) -> Result { + let invalid = |why: &str| RangeError::Invalid(format!("invalid Range {value:?}: {why}")); + let (unit, spec) = value + .split_once('=') + .ok_or_else(|| invalid("expected bytes=first-last"))?; + if !unit.trim().eq_ignore_ascii_case("bytes") { + return Err(invalid("only the bytes unit is supported")); + } + if spec.contains(',') { + return Err(invalid("multiple ranges are not supported")); + } + let (first, last) = spec + .trim() + .split_once('-') + .ok_or_else(|| invalid("expected bytes=first-last"))?; + // Digits only: `u64::from_str` also accepts a leading `+`. + let number = |s: &str| { + if s.is_empty() || !s.bytes().all(|b| b.is_ascii_digit()) { + return Err(invalid("positions must be decimal integers")); + } + s.parse::() + .map_err(|_| invalid("position out of range")) + }; + let (first, last) = (first.trim(), last.trim()); + + if first.is_empty() { + // Suffix range: the last `n` bytes. + let n = number(last)?; + if n == 0 || size == 0 { + return Err(RangeError::Unsatisfiable); + } + return Ok(ByteRange { + start: size.saturating_sub(n), + end: size, + }); + } + + let start = number(first)?; + let last = if last.is_empty() { + None + } else { + Some(number(last)?) + }; + if last.is_some_and(|last| last < start) { + return Err(invalid("last position is before first")); + } + if start >= size { + return Err(RangeError::Unsatisfiable); + } + let end = last.map_or(size, |last| last.saturating_add(1).min(size)); + Ok(ByteRange { start, end }) +} + +/// Build a range from the gRPC `offset` / `length` fields against an object of +/// `size` bytes. An absent `length` reads to the end and one past the end is +/// clamped; `length` 0 is [`RangeError::Invalid`] and an `offset` at or past +/// the end is [`RangeError::Unsatisfiable`]. +pub fn range_from_offset( + offset: u64, + length: Option, + size: u64, +) -> Result { + if length == Some(0) { + return Err(RangeError::Invalid("length must be positive".into())); + } + if offset >= size { + return Err(RangeError::Unsatisfiable); + } + let end = length.map_or(size, |length| offset.saturating_add(length).min(size)); + Ok(ByteRange { start: offset, end }) +} + +/// Split `range` into windows of at most `chunks_per_window` chunks, aligned to +/// chunk boundaries so adjacent windows never fetch the same chunk. +/// `chunk_sizes` are the plaintext chunk sizes in index order. +fn plan_windows(chunk_sizes: &[u64], range: ByteRange, chunks_per_window: usize) -> Vec { + let mut windows = Vec::new(); + let mut window_start = None; + let mut chunks_in_window = 0; + let mut chunk_start = 0u64; + for &chunk_size in chunk_sizes { + let chunk_end = chunk_start.saturating_add(chunk_size); + if chunk_start >= range.end { + break; + } + if chunk_end > range.start { + let start = *window_start.get_or_insert(chunk_start.max(range.start)); + chunks_in_window += 1; + if chunks_in_window == chunks_per_window { + windows.push(ByteRange { + start, + end: chunk_end.min(range.end), + }); + window_start = None; + chunks_in_window = 0; + } + } + chunk_start = chunk_end; + } + if let Some(start) = window_start { + windows.push(ByteRange { + start, + end: range.end, + }); + } + windows +} + +/// Stream `range` of a resolved root `data_map` as plaintext batches, one +/// window at a time. A failed window sends its error and ends the stream after +/// the bytes already delivered, which is the same contract as the whole-object +/// stream: a short body on REST, a terminal `Status` on gRPC. Dropping the +/// receiver stops the producer at the next window. +pub fn spawn_range_stream( + client: Arc, + data_map: DataMap, + range: ByteRange, +) -> mpsc::Receiver> { + let mut infos = data_map.infos().to_vec(); + infos.sort_by_key(|info| info.index); + let chunk_sizes: Vec = infos.iter().map(|info| info.src_size as u64).collect(); + let windows = plan_windows(&chunk_sizes, range, RANGE_WINDOW_CHUNKS); + + let (tx, rx) = mpsc::channel(1); + tokio::spawn(async move { + for window in windows { + let result = match (usize::try_from(window.start), usize::try_from(window.len())) { + (Ok(start), Ok(len)) => client.data_download_range(&data_map, start, len).await, + _ => Err(Error::InvalidData("range exceeds the address space".into())), + }; + let failed = result.is_err(); + if tx.send(result).await.is_err() || failed { + break; + } + } + }); + rx +} + +#[cfg(test)] +mod tests { + use super::*; + + fn r(start: u64, end: u64) -> ByteRange { + ByteRange { start, end } + } + + fn invalid(result: Result) -> bool { + matches!(result, Err(RangeError::Invalid(_))) + } + + #[test] + fn parses_first_last() { + assert_eq!(parse_range_header("bytes=0-99", 1000), Ok(r(0, 100))); + assert_eq!(parse_range_header("bytes=10-10", 1000), Ok(r(10, 11))); + assert_eq!(parse_range_header(" bytes = 5 - 9 ", 1000), Ok(r(5, 10))); + assert_eq!(parse_range_header("BYTES=5-9", 1000), Ok(r(5, 10))); + } + + #[test] + fn clamps_last_past_eof() { + assert_eq!(parse_range_header("bytes=900-5000", 1000), Ok(r(900, 1000))); + assert_eq!( + parse_range_header("bytes=0-18446744073709551615", 10), + Ok(r(0, 10)) + ); + } + + #[test] + fn parses_open_ended() { + assert_eq!(parse_range_header("bytes=990-", 1000), Ok(r(990, 1000))); + assert_eq!(parse_range_header("bytes=0-", 1), Ok(r(0, 1))); + } + + #[test] + fn parses_suffix() { + assert_eq!(parse_range_header("bytes=-100", 1000), Ok(r(900, 1000))); + assert_eq!(parse_range_header("bytes=-5000", 1000), Ok(r(0, 1000))); + } + + #[test] + fn start_at_or_past_eof_is_unsatisfiable() { + assert_eq!( + parse_range_header("bytes=1000-", 1000), + Err(RangeError::Unsatisfiable) + ); + assert_eq!( + parse_range_header("bytes=2000-3000", 1000), + Err(RangeError::Unsatisfiable) + ); + assert_eq!( + parse_range_header("bytes=-0", 1000), + Err(RangeError::Unsatisfiable) + ); + assert_eq!( + parse_range_header("bytes=0-", 0), + Err(RangeError::Unsatisfiable) + ); + assert_eq!( + parse_range_header("bytes=-1", 0), + Err(RangeError::Unsatisfiable) + ); + } + + #[test] + fn rejects_malformed() { + for value in [ + "", + "bytes", + "bytes=", + "bytes=-", + "bytes=abc-", + "bytes=+1-2", + "bytes=1-+2", + "bytes=1--2", + "bytes=10-5", + "items=0-1", + "bytes=0-1,5-6", + "bytes=99999999999999999999-", + ] { + assert!(invalid(parse_range_header(value, 1000)), "{value:?}"); + } + } + + #[test] + fn offset_length() { + assert_eq!(range_from_offset(0, None, 10), Ok(r(0, 10))); + assert_eq!(range_from_offset(4, Some(3), 10), Ok(r(4, 7))); + assert_eq!(range_from_offset(4, Some(100), 10), Ok(r(4, 10))); + assert_eq!(range_from_offset(9, Some(u64::MAX), 10), Ok(r(9, 10))); + assert_eq!( + range_from_offset(10, None, 10), + Err(RangeError::Unsatisfiable) + ); + assert_eq!( + range_from_offset(0, None, 0), + Err(RangeError::Unsatisfiable) + ); + assert!(invalid(range_from_offset(0, Some(0), 10))); + } + + #[test] + fn content_range_is_inclusive() { + assert_eq!(r(0, 100).content_range(1000), "bytes 0-99/1000"); + assert_eq!(r(999, 1000).content_range(1000), "bytes 999-999/1000"); + } + + #[test] + fn windows_within_one_chunk() { + assert_eq!( + plan_windows(&[100, 100, 100], r(110, 150), 8), + [r(110, 150)] + ); + } + + #[test] + fn windows_span_a_chunk_boundary() { + assert_eq!(plan_windows(&[100, 100, 100], r(90, 110), 8), [r(90, 110)]); + // One chunk per window: the boundary splits the range in two. + assert_eq!( + plan_windows(&[100, 100, 100], r(90, 110), 1), + [r(90, 100), r(100, 110)] + ); + } + + #[test] + fn windows_cover_the_last_partial_chunk() { + assert_eq!( + plan_windows(&[100, 100, 37], r(150, 237), 1), + [r(150, 200), r(200, 237)] + ); + } + + #[test] + fn windows_are_chunk_aligned_and_cover_the_range_exactly() { + let sizes = [100u64; 20]; + let range = r(250, 1730); + let windows = plan_windows(&sizes, range, 4); + // 250..1730 overlaps chunks 2..=17 (16 chunks): 4 windows of 4. + assert_eq!( + windows, + [r(250, 600), r(600, 1000), r(1000, 1400), r(1400, 1730)] + ); + // Contiguous, and every interior boundary is a chunk boundary. + assert_eq!(windows.first().unwrap().start, range.start); + assert_eq!(windows.last().unwrap().end, range.end); + for pair in windows.windows(2) { + assert_eq!(pair[0].end, pair[1].start); + assert_eq!(pair[0].end % 100, 0); + } + } + + #[test] + fn whole_object_window() { + assert_eq!(plan_windows(&[100, 100, 37], r(0, 237), 8), [r(0, 237)]); + } +} diff --git a/antd/src/rest/data.rs b/antd/src/rest/data.rs index aec32458..d561e2e1 100644 --- a/antd/src/rest/data.rs +++ b/antd/src/rest/data.rs @@ -2,7 +2,7 @@ use std::sync::Arc; use axum::body::Body; use axum::extract::{Path, State}; -use axum::http::{header, HeaderMap, StatusCode}; +use axum::http::{header, HeaderMap, HeaderValue, StatusCode}; use axum::response::Response; use axum::Json; use base64::engine::general_purpose::STANDARD as BASE64; @@ -285,10 +285,56 @@ async fn stream_response( .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") .header(header::CONTENT_LENGTH, content_length.to_string()) + .header(header::ACCEPT_RANGES, "bytes") .body(body) .expect("static content-type + numeric content-length are always valid")) } +/// Build a `206 Partial Content` response for a `Range` request (V2-1425): +/// stream only the requested plaintext bytes, fetching and decrypting just the +/// chunks that overlap them (see [`crate::range`]). One `bytes` range is +/// supported. A range starting at or past the end is `416` with +/// `Content-Range: bytes */size`; a malformed or multi-range header is `400`. +/// As on the whole-object path, `Content-Length` is the promised byte count, so +/// a body that ends short signals a failed download. +/// +/// Ranges are raw-only: combined with the NDJSON progress framing they are +/// rejected as `400`. +async fn range_response( + client: Arc, + data_map: ant_core::data::DataMap, + range_header: &HeaderValue, + ndjson: bool, +) -> Result { + if ndjson { + return Err(AntdError::BadRequest(format!( + "Range is not supported with {NDJSON_CONTENT_TYPE} progress framing" + ))); + } + let range_header = range_header + .to_str() + .map_err(|_| AntdError::BadRequest("Range header is not valid ASCII".into()))?; + let data_map = crate::datamap::resolve_root_data_map(&client, data_map) + .await + .map_err(AntdError::from_core)?; + let size = data_map.original_file_size() as u64; + let range = crate::range::parse_range_header(range_header, size).map_err(|e| match e { + crate::range::RangeError::Invalid(msg) => AntdError::BadRequest(msg), + crate::range::RangeError::Unsatisfiable => AntdError::RangeNotSatisfiable { size }, + })?; + + let rx = crate::range::spawn_range_stream(client, data_map, range); + let body = Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx)); + Ok(Response::builder() + .status(StatusCode::PARTIAL_CONTENT) + .header(header::CONTENT_TYPE, "application/octet-stream") + .header(header::CONTENT_LENGTH, range.len().to_string()) + .header(header::CONTENT_RANGE, range.content_range(size)) + .header(header::ACCEPT_RANGES, "bytes") + .body(body) + .expect("static content-type + numeric range headers are always valid")) +} + /// Build the opt-in NDJSON streaming response: interleave fetch-progress frames /// with the data frames so a consumer can drive a *determinate* progress bar /// (the raw path's byte delivery is lumpy — often the whole file in one batch — @@ -384,7 +430,9 @@ async fn stream_response_ndjson( /// `POST /v1/data/stream` — private streaming download from a caller-held /// DataMap (the primitive). Mirrors `data_get` but streams the plaintext /// instead of buffering it into a base64 JSON body. POST (not GET) because the -/// hex DataMap can be many KB. +/// hex DataMap can be many KB. A `Range` header returns just that byte range +/// (see [`range_response`]); RFC 9110 defines `Range` only for GET, so this is +/// an antd extension on this route. pub async fn data_stream( State(state): State>, headers: HeaderMap, @@ -407,7 +455,15 @@ pub async fn data_stream( let data_map: ant_core::data::DataMap = rmp_serde::from_slice(&data_map_bytes) .map_err(|e| AntdError::BadRequest(format!("invalid data map: {e}")))?; - if wants_ndjson(&headers) { + if let Some(range) = headers.get(header::RANGE) { + range_response( + state.client.clone(), + data_map, + range, + wants_ndjson(&headers), + ) + .await + } else if wants_ndjson(&headers) { stream_response_ndjson(state.client.clone(), data_map).await } else { stream_response(state.client.clone(), data_map).await @@ -417,7 +473,7 @@ pub async fn data_stream( /// `GET /v1/data/public/{addr}/stream` — public streaming download. Resolves /// the address to a DataMap, then streams from it (wraps the private /// primitive). A fetch failure surfaces as a normal error response before the -/// stream opens. +/// stream opens. Honours a `Range` header like `data_stream`. pub async fn data_stream_public( State(state): State>, headers: HeaderMap, @@ -440,7 +496,15 @@ pub async fn data_stream_public( .await .map_err(AntdError::from_core)?; - if wants_ndjson(&headers) { + if let Some(range) = headers.get(header::RANGE) { + range_response( + state.client.clone(), + data_map, + range, + wants_ndjson(&headers), + ) + .await + } else if wants_ndjson(&headers) { stream_response_ndjson(state.client.clone(), data_map).await } else { stream_response(state.client.clone(), data_map).await