diff --git a/documentation/docs/learn_more/applications.mdx b/documentation/docs/learn_more/applications.mdx index 0550d040..92cac60c 100644 --- a/documentation/docs/learn_more/applications.mdx +++ b/documentation/docs/learn_more/applications.mdx @@ -13,19 +13,19 @@ Your entrypoint will construct an `Application` and then `run` it: +(CODE:src=../../../reboot/examples/chat-room/backend/src/main.py&lines=26-34) --> ```py async def main(): await Application( servicers=[ChatRoomServicer], + # Message attachments are stored as blobs; `rbt dev run` and + # `rbt serve run` provide a local filesystem data plane (see + # `REBOOT_BLOB_DATA_PLANE_URL` for using a custom one). + libraries=[blobs_library()], initialize=initialize, ).run() - - -if __name__ == '__main__': - asyncio.run(main()) ``` @@ -59,7 +59,7 @@ instances used by your application are created. +(CODE:src=../../../reboot/examples/chat-room/backend/src/main.py&lines=14-21) --> ```py @@ -129,7 +129,7 @@ implemented with [Express.js](https://expressjs.com/). **Limitations of custom HTTP routes** -* Currently only `GET` and `POST` methods are supported. +* Currently only `GET`, `POST`, and `PUT` methods are supported. * The `/` route is currently used by Reboot itself to show a helpful page explaining that this is a Reboot diff --git a/documentation/docs/learn_more/call/from_react.mdx b/documentation/docs/learn_more/call/from_react.mdx index a9897c8e..4bbdc9a4 100644 --- a/documentation/docs/learn_more/call/from_react.mdx +++ b/documentation/docs/learn_more/call/from_react.mdx @@ -90,14 +90,33 @@ api = API( +(CODE:src=../../../../reboot/examples/chat-room/api/chat_room/v1/chat_room.proto&&lines=9-51&syntax=protobuf) --> ```protobuf +// Raised when a requested attachment exceeds the chat room's +// per-attachment size limit. +message AttachmentTooLarge { + // The maximum allowed attachment size, in bytes, so the client can + // show a helpful message. + uint64 max_size_bytes = 1; +} + +message Message { + string text = 1; + + // Ids of `rbt.std.blobs.v1.Blob`s holding this message's + // attachments. Attachments are visible to all participants from the + // moment the message is sent, while their bytes may still be + // uploading; render their progress reactively via `Blob.Info`. + repeated string attachment_blob_ids = 2; +} + message ChatRoom { option (rbt.v1alpha1.state) = { }; - repeated string messages = 1; + reserved 1; + repeated Message messages = 2; } service ChatRoomMethods { @@ -107,24 +126,17 @@ service ChatRoomMethods { }; } - // Adds a new message to the list of recorded messages. + // Adds a new message to the list of recorded messages, creating a + // `Blob` for each requested attachment. The message is published + // immediately; the caller then uploads the attachment bytes into + // the returned blob ids. rpc Send(SendRequest) returns (SendResponse) { - option (rbt.v1alpha1.method).writer = { + option (rbt.v1alpha1.method) = { + transaction: {}, + errors: [ "AttachmentTooLarge" ], }; } } - -message MessagesRequest {} - -message MessagesResponse { - repeated string messages = 1; -} - -message SendRequest { - string message = 1; // E.g. "Hello, World". -} - -message SendResponse {} ``` @@ -266,7 +278,7 @@ export const DashboardApp: FC = ({ You can call a `reader` _reactively_ very simply: +(CODE:src=../../../../reboot/examples/chat-room/frontend/web/src/App.tsx&&lines=156-157) --> ```tsx @@ -295,7 +307,7 @@ and [`transaction`](/learn_more/define/methods#kinds) methods are both callable from React. +(CODE:src=../../../../reboot/examples/chat-room/frontend/web/src/App.tsx&&lines=156-156) --> ```tsx @@ -312,11 +324,17 @@ This line calls the in the `.proto` as `Send`, using the lower camel case name `send`: +(CODE:src=../../../../reboot/examples/chat-room/frontend/web/src/App.tsx&&lines=177-183) --> ```tsx -const { aborted } = await send({ message: message }); +const { response, aborted } = await send({ + message: message, + attachments: + file !== null + ? [{ contentType: file.type, sizeBytes: BigInt(file.size) }] + : [], +}); ``` @@ -333,12 +351,17 @@ Reboot attaches all in-flight mutations to a `.pending` property of every mutato facilitate this, for example: +(CODE:src=../../../../reboot/examples/chat-room/frontend/web/src/App.tsx&&lines=302-309) --> ```tsx -{send.pending.map(({ request: { message }, isLoading }) => ( - +{send.pending.map(({ request, isLoading }, index) => ( + ))} ``` @@ -365,15 +388,30 @@ Every call to a mutator returns both a `response` and an `aborted`; successful c [Learn more about Reboot errors and error types.](/learn_more/errors) +(CODE:src=../../../../reboot/examples/chat-room/frontend/web/src/App.tsx&&lines=177-196) --> ```tsx -const { aborted } = await send({ message: message }); +const { response, aborted } = await send({ + message: message, + attachments: + file !== null + ? [{ contentType: file.type, sizeBytes: BigInt(file.size) }] + : [], +}); if (aborted !== undefined) { - console.warn(aborted.error.getType()); - console.warn(aborted.message); -} + // Surface the failure to the user. The backend's + // `AttachmentTooLarge` error carries the limit, so we can say + // exactly what it is. + if (aborted.error instanceof AttachmentTooLarge) { + const maxMebibytes = Number(aborted.error.maxSizeBytes) / (1024 * 1024); + setError( + `That attachment is too large. The maximum is ${maxMebibytes} MiB.` + ); + } else { + setError(`Couldn't send your message: ${aborted.message}`); + } + return; ``` diff --git a/documentation/docs/learn_more/call/from_within_your_app.mdx b/documentation/docs/learn_more/call/from_within_your_app.mdx index 24ca2f9c..024a1efc 100644 --- a/documentation/docs/learn_more/call/from_within_your_app.mdx +++ b/documentation/docs/learn_more/call/from_within_your_app.mdx @@ -83,14 +83,15 @@ passed as a parameter to your method. ```py + self, + context: ReaderContext, + request: MessagesRequest, +) -> MessagesResponse: + return MessagesResponse(messages=self.state.messages) + async def send( self, - context: WriterContext, - request: SendRequest, -) -> SendResponse: - message = request.message - self.state.messages.extend([message]) - return SendResponse() + context: TransactionContext, ``` diff --git a/documentation/docs/learn_more/define/protobuf.mdx b/documentation/docs/learn_more/define/protobuf.mdx index 23481476..38ccfe8e 100644 --- a/documentation/docs/learn_more/define/protobuf.mdx +++ b/documentation/docs/learn_more/define/protobuf.mdx @@ -10,14 +10,15 @@ To declare a data type called `ChatRoom`, put the following in a `.proto` file: +(CODE:src=../../../../reboot/examples/chat-room/api/chat_room/v1/chat_room.proto&lines=27-32&syntax=protobuf) --> ```protobuf message ChatRoom { option (rbt.v1alpha1.state) = { }; - repeated string messages = 1; + reserved 1; + repeated Message messages = 2; } ``` @@ -32,7 +33,7 @@ In addition to defining the data type, you'll also need to define operations for that type, for example: +(CODE:src=../../../../reboot/examples/chat-room/api/chat_room/v1/chat_room.proto&lines=34-51&syntax=protobuf) --> ```protobuf @@ -43,9 +44,14 @@ service ChatRoomMethods { }; } - // Adds a new message to the list of recorded messages. + // Adds a new message to the list of recorded messages, creating a + // `Blob` for each requested attachment. The message is published + // immediately; the caller then uploads the attachment bytes into + // the returned blob ids. rpc Send(SendRequest) returns (SendResponse) { - option (rbt.v1alpha1.method).writer = { + option (rbt.v1alpha1.method) = { + transaction: {}, + errors: [ "AttachmentTooLarge" ], }; } } diff --git a/rbt/std/BUILD.bazel b/rbt/std/BUILD.bazel index f2ba5d5e..6f1b2481 100644 --- a/rbt/std/BUILD.bazel +++ b/rbt/std/BUILD.bazel @@ -47,6 +47,9 @@ ts_project( }, visibility = ["//visibility:public"], deps = [ + "//rbt/std/blobs/v1:blobs_js_proto", + "//rbt/std/blobs/v1:blobs_js_reboot", + "//rbt/std/blobs/v1:blobs_js_reboot_react", "//rbt/std/ciphertext/v1:ciphertext_js_proto", "//rbt/std/ciphertext/v1:ciphertext_js_reboot", "//rbt/std/collections/ordered_map/v1:ordered_map_js_proto", diff --git a/rbt/std/blobs/v1/BUILD.bazel b/rbt/std/blobs/v1/BUILD.bazel new file mode 100644 index 00000000..b108cc43 --- /dev/null +++ b/rbt/std/blobs/v1/BUILD.bazel @@ -0,0 +1,92 @@ +load( + "@com_github_reboot_dev_reboot//reboot:rules.bzl", + "js_proto_library", + "js_reboot_library", + "js_reboot_react_library", + "py_reboot_library", +) +load("@com_google_protobuf//bazel:proto_library.bzl", "proto_library") + +proto_library( + name = "blobs_proto", + srcs = [ + ":blobs.proto", + ], + visibility = ["//visibility:public"], + deps = [ + "@com_github_reboot_dev_reboot//rbt/v1alpha1:options_proto", + "@com_google_protobuf//:descriptor_proto", + ], +) + +# The blob data-plane interface: a plain gRPC service (no Reboot state +# options). Built with `py_reboot_library` — which, besides the plain +# `_pb2`/`_pb2_grpc` modules the filesystem server uses directly, emits +# the `_rbt` module that lets the Cloud facilitator register it via +# `legacy_grpc_servicers`. Python-only: the JS SDK talks to the `Blob` +# control plane, never the data plane directly. +proto_library( + name = "data_plane_proto", + srcs = [ + ":data_plane.proto", + ], + visibility = ["//visibility:public"], +) + +py_reboot_library( + name = "data_plane_py_reboot", + proto = "data_plane.proto", + proto_library = ":data_plane_proto", + visibility = ["//visibility:public"], +) + +py_reboot_library( + name = "blobs_py_reboot", + proto = "blobs.proto", + proto_library = ":blobs_proto", + visibility = ["//visibility:public"], +) + +js_proto_library( + name = "blobs_js_proto", + package_json = ":package.json", + proto = "blobs.proto", + proto_deps = [ + ":blobs_proto", + # ISSUE(https://github.com/reboot-dev/mono/issues/3218): Until we can + # use `create_protoc_plugin_rule` we need to repeat the dependencies of + # the `proto_libraries` here. + "@com_github_reboot_dev_reboot//rbt/v1alpha1:options_proto", + "@com_google_protobuf//:descriptor_proto", + ], + visibility = ["//visibility:public"], +) + +js_reboot_library( + name = "blobs_js_reboot", + srcs = [ + ":blobs_proto", + ], + proto = "blobs.proto", + visibility = ["//visibility:public"], + deps = [ + ":blobs_js_proto", + ], +) + +js_reboot_react_library( + name = "blobs_js_reboot_react", + srcs = [ + ":blobs_js_proto", + ], + proto = "blobs.proto", + proto_deps = [ + ":blobs_proto", + # ISSUE(https://github.com/reboot-dev/mono/issues/3218): Until we can + # use `create_protoc_plugin_rule` we need to repeat the dependencies of + # the `proto_libraries` here. + "@com_github_reboot_dev_reboot//rbt/v1alpha1:options_proto", + "@com_google_protobuf//:descriptor_proto", + ], + visibility = ["//visibility:public"], +) diff --git a/rbt/std/blobs/v1/blobs.proto b/rbt/std/blobs/v1/blobs.proto new file mode 100644 index 00000000..045a0094 --- /dev/null +++ b/rbt/std/blobs/v1/blobs.proto @@ -0,0 +1,414 @@ +syntax = "proto3"; + +package rbt.std.blobs.v1; + +import "rbt/v1alpha1/options.proto"; + +//////////////////////////////////////////////////////////////////////// +// Errors. + +// Raised when an upload-side call arrives after the blob has already +// been committed (or is in the process of being deleted). +message AlreadyCommitted {} + +// Raised when `DownloadUrl` is called before the blob has been +// committed. +message NotCommitted {} + +// Raised when the uploaded parts violate the blob's declared `size` +// or `max_size`. +message SizeMismatch { + // The total size of all parts reported uploaded so far. + uint64 bytes_uploaded = 1; +} + +// Raised when `Commit` is called with no parts uploaded, or with a +// set of part numbers that is not contiguous starting at 1. +message IncompleteParts {} + +//////////////////////////////////////////////////////////////////////// + +// The set of users allowed to download a blob. Wrapping the list in a +// message lets the download allow-list be *omitted* (anyone who knows +// the blob's id may download) distinctly from *present but empty* (no +// one but app-internal callers may download). +message Downloaders { + repeated string user_ids = 1; +} + +//////////////////////////////////////////////////////////////////////// +// Blob: the control plane for one immutable-once-committed binary +// object. This state holds only *metadata*; the bytes live in a +// `BlobDataPlane` (see `data_plane.proto` — e.g. the local filesystem +// server, or an S3-backed service) and travel directly between the +// client and that data plane via URLs minted by `UploadInstructions` +// and `DownloadUrl`. The upload protocol follows S3 multipart +// semantics: numbered parts PUT to per-part URLs, each returning an +// ETag, then a completion step that validates those ETags. + +message BlobPart { + // 1-based part number, following S3 multipart numbering. + uint32 number = 1; + + // The ETag the data plane returned when this part was uploaded. + string etag = 2; + + // Size of this part in bytes. + uint64 size = 3; +} + +message Blob { + option (rbt.v1alpha1.state) = { + trusted_effects: true, + }; + + enum Status { + // Parts may be uploaded. The initial status. + UPLOADING = 0; + + // `Commit` was called; the `CompleteUpload` workflow is + // finalizing the object in the storage backend. + COMMITTING = 1; + + // The blob's bytes are immutable and downloadable. + COMMITTED = 2; + + // `Remove` was called (or the upload expired); the + // `PerformRemove` workflow is removing the bytes from the + // storage backend. + DELETING = 3; + + // The bytes are gone; only this metadata tombstone remains. + DELETED = 4; + } + + Status status = 1; + + // MIME type served on download, e.g. `image/png`. + string content_type = 2; + + // Expected total size in bytes, when known at `Create` time; + // enables definite progress tracking and is enforced at `Commit`. + optional uint64 size = 3; + + // Upper bound on the total size in bytes, enforced as parts are + // reported and at `Commit`. + optional uint64 max_size = 4; + + // The user that may upload into this blob. Empty means anyone who + // knows this blob's id may upload; see the authorizer note in + // `reboot.std.blobs.v1.blobs`. + string uploader_id = 5; + + // Who may download this blob. When unset (omitted at `Create`), + // anyone who knows the blob's id may download. When set, only the + // listed users may (an empty list means no one but app-internal + // callers). The uploader is NOT implicitly a downloader. See the + // authorizer note in `reboot.std.blobs.v1.blobs`. + optional Downloaders downloader_ids = 10; + + // Storage-backend upload session id, set by the `BeginUpload` + // workflow. `UploadInstructions` reports `ready: false` until this + // is set. + optional string upload_id = 6; + + // The parts reported uploaded so far, ordered by part number. + repeated BlobPart parts = 7; + + // Composite ETag of the committed object. + optional string etag = 8; + + // Why the most recent `CompleteUpload` attempt failed. Cleared on + // success; while set, `status` has reverted to `UPLOADING` so the + // client can re-upload parts and `Commit` again. The message + // describes the failure; it does not identify which parts, if any, + // were at fault. + optional string commit_error = 9; +} + +//////////////////////////////////////////////////////////////////////// + +message CreateRequest { + // MIME type served on download, e.g. `image/png`. + string content_type = 1; + + // Expected exact total size in bytes, if known. Enables definite + // progress bars, and is enforced at `Commit`: the committed bytes + // must total exactly `size`. Optional and independent of + // `max_size`; see `max_size` for how the two interact. + optional uint64 size = 2; + + // Upper bound on the total size in bytes. Enforced as parts are + // reported (via `PartUploaded`) and again at `Commit`: the running + // total must never exceed `max_size`. `size` and `max_size` are + // each optional and may be set together or independently. When + // both are set both constraints apply, so a sensible caller keeps + // `size <= max_size`. + optional uint64 max_size = 3; + + // The user that may upload into this blob. Leave empty to allow + // anyone who knows this blob's id to upload. + string uploader_id = 4; + + // Who may download this blob. Omit the field entirely to let anyone + // who knows the blob's id download it; set it (even to an empty + // list) to restrict downloads to the listed users. The uploader is + // NOT implicitly a downloader — omit them from the list if they + // shouldn't be able to download (e.g. a drop-box). Can be changed + // later with `SetDownloaders`. + optional Downloaders downloader_ids = 5; +} + +message CreateResponse {} + +//////////////////////////////////////////////////////////////////////// + +message SetDownloadersRequest { + // The new download allow-list. Omit the field entirely to remove + // any restriction (anyone who knows the blob's id may download); + // set it (even to an empty list) to restrict downloads to the + // listed users. + optional Downloaders downloader_ids = 1; +} + +message SetDownloadersResponse {} + +//////////////////////////////////////////////////////////////////////// + +message BeginUploadRequest {} + +message BeginUploadResponse {} + +//////////////////////////////////////////////////////////////////////// + +message UploadInstructionsRequest { + // The 1-based part numbers to mint upload URLs for. + repeated uint32 part_numbers = 1; +} + +message PartUploadInstruction { + uint32 part_number = 1; + + // URL to `PUT` this part's bytes to. May be relative (resolve it + // against the application's URL) or absolute (e.g. a presigned S3 + // URL). The response's `ETag` header must be reported back via + // `PartUploaded`. + string url = 2; +} + +message UploadInstructionsResponse { + // False until the `BeginUpload` workflow has provisioned the + // storage-backend upload session; call this method reactively (or + // poll) until true. + bool ready = 1; + + // The part size uploads should use: every part except the last + // must be exactly this many bytes. + uint64 part_size = 2; + + repeated PartUploadInstruction instructions = 3; +} + +//////////////////////////////////////////////////////////////////////// + +message PartUploadedRequest { + uint32 part_number = 1; + + // The ETag returned by the data plane for this part. Validated by + // the storage backend at `Commit` time. + string etag = 2; + + // Size of this part in bytes. + uint64 size = 3; +} + +message PartUploadedResponse {} + +//////////////////////////////////////////////////////////////////////// + +message CommitRequest {} + +message CommitResponse {} + +//////////////////////////////////////////////////////////////////////// + +message CompleteUploadRequest {} + +message CompleteUploadResponse {} + +//////////////////////////////////////////////////////////////////////// + +message InfoRequest {} + +message InfoResponse { + Blob.Status status = 1; + string content_type = 2; + optional uint64 size = 3; + optional uint64 max_size = 4; + string uploader_id = 5; + + // Total bytes across the parts reported uploaded so far. Together + // with `size` this gives upload progress, observable reactively by + // any client authorized to call `Info`. + uint64 bytes_uploaded = 6; + + repeated BlobPart parts = 7; + optional string etag = 8; + optional string commit_error = 9; +} + +//////////////////////////////////////////////////////////////////////// + +message DownloadUrlRequest { + // How long the minted URL should remain valid. The storage + // backend's default (and maximum) applies when unset. + optional uint32 ttl_seconds = 1; +} + +message DownloadUrlResponse { + // URL to `GET` the blob's bytes from. May be relative (resolve it + // against the application's URL) or absolute (e.g. a presigned S3 + // URL). + string url = 1; +} + +//////////////////////////////////////////////////////////////////////// + +message RemoveRequest {} + +message RemoveResponse {} + +//////////////////////////////////////////////////////////////////////// + +message PerformRemoveRequest {} + +message PerformRemoveResponse {} + +//////////////////////////////////////////////////////////////////////// + +message ExpireIfNotCommittedRequest {} + +message ExpireIfNotCommittedResponse {} + +//////////////////////////////////////////////////////////////////////// + +service BlobMethods { + // Creates the blob's metadata and schedules the `BeginUpload` + // workflow that provisions an upload session in the storage + // backend. Only application code may call this; it is the + // application's chance to enforce quota and size policy (directly, + // or by setting `max_size`), to record which user may upload + // (`uploader_id`), and to restrict who may download (`downloader_ids`). + rpc Create(CreateRequest) returns (CreateResponse) { + option (rbt.v1alpha1.method) = { + writer: { constructor: {} }, + }; + } + + // Replaces the blob's download allow-list. Application-mediated + // (app-internal only), like `Create`. Omit `downloader_ids` in the + // request to remove any restriction (anyone who knows the id may + // download again); set it to restrict downloads to the listed + // users. + rpc SetDownloaders(SetDownloadersRequest) returns (SetDownloadersResponse) { + option (rbt.v1alpha1.method) = { + writer: {}, + }; + } + + // Provisions the storage-backend upload session and records its + // `upload_id`. Scheduled by `Create`; not for direct use. + rpc BeginUpload(BeginUploadRequest) returns (BeginUploadResponse) { + option (rbt.v1alpha1.method) = { + workflow: {}, + }; + } + + // Mints `PUT` URLs for the requested part numbers. Reports + // `ready: false` until `BeginUpload` has completed. + rpc UploadInstructions(UploadInstructionsRequest) + returns (UploadInstructionsResponse) { + option (rbt.v1alpha1.method) = { + reader: {}, + errors: [ "AlreadyCommitted" ], + }; + } + + // Records that a part was uploaded to the data plane, updating the + // progress observable via `Info`. Safe to call multiple times for the + // same part number; a re-uploaded part overwrites its previous + // record. + rpc PartUploaded(PartUploadedRequest) returns (PartUploadedResponse) { + option (rbt.v1alpha1.method) = { + writer: {}, + errors: [ "AlreadyCommitted", "SizeMismatch", "IncompleteParts" ], + }; + } + + // Validates the reported parts and schedules the `CompleteUpload` + // workflow that finalizes the object in the storage backend. + // Observe the outcome reactively via `Info`: `status` becomes + // `COMMITTED`, or reverts to `UPLOADING` with `commit_error` set. + rpc Commit(CommitRequest) returns (CommitResponse) { + option (rbt.v1alpha1.method) = { + writer: {}, + errors: [ "AlreadyCommitted", "SizeMismatch", "IncompleteParts" ], + }; + } + + // Finalizes the object in the storage backend (which validates the + // reported part ETags). Scheduled by `Commit`; not for direct use. + rpc CompleteUpload(CompleteUploadRequest) returns (CompleteUploadResponse) { + option (rbt.v1alpha1.method) = { + workflow: {}, + }; + } + + // Metadata and upload progress. Reactive: watch it to render a + // progress bar. Visible to anyone who may upload or download the + // blob: the `uploader_id` and any listed `downloader_ids`, plus anyone + // who knows the id whenever either side is left open. + rpc Info(InfoRequest) returns (InfoResponse) { + option (rbt.v1alpha1.method) = { + reader: {}, + }; + } + + // Mints a time-limited URL from which the committed blob's bytes can + // be downloaded. + rpc DownloadUrl(DownloadUrlRequest) returns (DownloadUrlResponse) { + option (rbt.v1alpha1.method) = { + reader: {}, + errors: [ "NotCommitted" ], + }; + } + + // Marks the blob for deletion and schedules the `PerformRemove` + // workflow that removes its bytes from the storage backend. + // (Named `Remove` because `Delete` is a reserved Reboot method + // name.) + rpc Remove(RemoveRequest) returns (RemoveResponse) { + option (rbt.v1alpha1.method) = { + writer: {}, + }; + } + + // Removes the blob's bytes from the storage backend. Scheduled by + // `Remove` and `ExpireIfNotCommitted`; not for direct use. + rpc PerformRemove(PerformRemoveRequest) returns (PerformRemoveResponse) { + option (rbt.v1alpha1.method) = { + workflow: {}, + }; + } + + // Expunges the blob if its upload was abandoned: scheduled by + // `Create` to run after the upload expiration period, it deletes + // the blob unless it has been committed by then. Not for direct + // use. + rpc ExpireIfNotCommitted(ExpireIfNotCommittedRequest) + returns (ExpireIfNotCommittedResponse) { + option (rbt.v1alpha1.method) = { + writer: {}, + }; + } +} diff --git a/rbt/std/blobs/v1/data_plane.proto b/rbt/std/blobs/v1/data_plane.proto new file mode 100644 index 00000000..2d5eb172 --- /dev/null +++ b/rbt/std/blobs/v1/data_plane.proto @@ -0,0 +1,186 @@ +// The blob data-plane interface. +// +// A `BlobDataPlane` holds blob *bytes*; all blob *metadata* lives in +// the `Blob` control-plane state machine (see `blobs.proto`). The +// control plane never touches bytes: it calls this service to +// provision uploads, finalize them, mint download URLs, and delete +// objects, and the bytes themselves travel directly between the client +// and wherever the data plane keeps them (a presigned-URL store like +// S3, or — for a data plane whose `Configuration` asks for forwarded +// paths — the application's own HTTP routes proxying to it). +// +// This is a plain gRPC service, deliberately free of any Reboot +// framework options, so that it can be implemented by anything and +// addressed by a bare URL (`REBOOT_BLOB_DATA_PLANE_URL`). The +// open-source filesystem server implements it; Reboot Cloud implements +// it privately (S3, and eventually a CDN) in each app's facilitator. +// +// The control plane retries `BeginUpload`, `CompleteUpload`, and +// `Delete` inside workflows, so every method must be *safe* to call +// more than once for the same `blob_id`. `CompleteUpload` and `Delete` +// are naturally idempotent (a completed blob returns its ETag; +// deleting an absent blob succeeds). `BeginUpload` should reuse an +// existing uncommitted session where it can; a backend that can only +// mint fresh upload ids is still acceptable — a lost response then +// orphans a session, which is reclaimed when the blob expires +// uncommitted or is deleted. + +syntax = "proto3"; + +package rbt.std.blobs.v1; + +service BlobDataPlane { + // Reports what the control plane needs to know to talk to this data + // plane: the client part size, and which paths (if any) the + // application must forward to it. Read once at application startup. + rpc Configuration(ConfigurationRequest) returns (ConfigurationResponse); + + // Provisions an upload session for a blob and returns its data-plane + // upload id (e.g. an S3 multipart upload id). Should reuse an + // existing uncommitted session for the same blob; a fresh session is + // acceptable (see the idempotency note above). + rpc BeginUpload(DataPlaneBeginUploadRequest) + returns (DataPlaneBeginUploadResponse); + + // Mints URLs to which the client `PUT`s each requested part's bytes + // directly. A URL is either absolute (e.g. a presigned S3 URL, + // directly reachable by the client) or application-relative, under a + // path the data plane's `Configuration` asked to have forwarded. + rpc UploadInstructions(DataPlaneUploadInstructionsRequest) + returns (DataPlaneUploadInstructionsResponse); + + // Finalizes the object from its uploaded parts, validating the + // reported ETags/sizes against the real bytes and enforcing + // `max_size` against the real total. Returns the composite ETag, or + // a permanent-failure `error` (ETag mismatch, over `max_size`, ...) + // that the control plane surfaces so the client can re-upload. A + // *transient* failure is a gRPC error instead, so the control-plane + // workflow retries. Idempotent: completing an already-completed blob + // returns its ETag. + rpc CompleteUpload(DataPlaneCompleteUploadRequest) + returns (DataPlaneCompleteUploadResponse); + + // Mints a URL from which the committed object's bytes can be + // downloaded: absolute (e.g. a presigned S3/CloudFront URL) or + // application-relative under a forwarded path, like + // `UploadInstructions` URLs. Minted afresh per request so that + // swapping the data-plane implementation takes effect immediately. + rpc GetDownloadUrl(DataPlaneGetDownloadUrlRequest) + returns (DataPlaneGetDownloadUrlResponse); + + // Removes the object's bytes and any incomplete upload session. + // Idempotent: deleting an absent blob succeeds. + rpc Delete(DataPlaneDeleteRequest) returns (DataPlaneDeleteResponse); +} + +message ConfigurationRequest {} + +// The HTTP method of a forwarded path. +enum HttpMethod { + HTTP_METHOD_UNSPECIFIED = 0; + HTTP_METHOD_GET = 1; + HTTP_METHOD_PUT = 2; +} + +// One path namespace the application must forward to the data plane: +// requests with the given method whose path starts with `path_prefix` +// are reverse-proxied to the data plane's HTTP endpoint verbatim +// (path and query). `path_prefix` must be under `/__/reboot/blob/` — +// the application refuses anything else, so a data plane can never +// claim application routes. +message ForwardedPath { + HttpMethod method = 1; + string path_prefix = 2; +} + +message ConfigurationResponse { + // The part size clients must use; every part but the last must be + // exactly this size. + uint64 part_size = 1; + + // Paths the application must forward to this data plane. Empty when + // the data plane's URLs are directly reachable by clients (e.g. + // presigned S3 or CloudFront URLs) and no forwarding is needed. + repeated ForwardedPath forwarded_paths = 2; + + // The port of this data plane's HTTP byte endpoint, to which + // forwarded requests are proxied. The host is the one the + // application already reaches this gRPC service on (so the data + // plane need not know how it is addressed externally), and the + // scheme follows that connection's transport security. Only + // meaningful when `forwarded_paths` is non-empty. + uint32 http_port = 3; +} + +//////////////////////////////////////////////////////////////////////// + +message DataPlaneBeginUploadRequest { + string blob_id = 1; + string content_type = 2; +} + +message DataPlaneBeginUploadResponse { + string upload_id = 1; +} + +//////////////////////////////////////////////////////////////////////// + +message DataPlaneUploadInstructionsRequest { + string blob_id = 1; + string upload_id = 2; + repeated uint32 part_numbers = 3; +} + +message DataPlanePartUploadInstruction { + uint32 part_number = 1; + string url = 2; +} + +message DataPlaneUploadInstructionsResponse { + repeated DataPlanePartUploadInstruction instructions = 1; +} + +//////////////////////////////////////////////////////////////////////// + +message DataPlaneUploadedPart { + uint32 number = 1; + string etag = 2; + uint64 size = 3; +} + +message DataPlaneCompleteUploadRequest { + string blob_id = 1; + string upload_id = 2; + string content_type = 3; + repeated DataPlaneUploadedPart parts = 4; + optional uint64 max_size = 5; +} + +message DataPlaneCompleteUploadResponse { + // The committed object's composite ETag. Empty when `error` is set. + string etag = 1; + + // A permanent-failure reason, if completion failed in a way the + // client can fix by re-uploading. When set, `etag` is empty and the + // control plane reverts the blob to UPLOADING with this message. + optional string error = 2; +} + +//////////////////////////////////////////////////////////////////////// + +message DataPlaneGetDownloadUrlRequest { + string blob_id = 1; + optional uint64 ttl_seconds = 2; +} + +message DataPlaneGetDownloadUrlResponse { + string url = 1; +} + +//////////////////////////////////////////////////////////////////////// + +message DataPlaneDeleteRequest { + string blob_id = 1; +} + +message DataPlaneDeleteResponse {} diff --git a/rbt/std/blobs/v1/package.json b/rbt/std/blobs/v1/package.json new file mode 100644 index 00000000..3dbc1ca5 --- /dev/null +++ b/rbt/std/blobs/v1/package.json @@ -0,0 +1,3 @@ +{ + "type": "module" +} diff --git a/reboot/BUILD.bazel b/reboot/BUILD.bazel index 2bb7b4cd..74b73959 100644 --- a/reboot/BUILD.bazel +++ b/reboot/BUILD.bazel @@ -473,6 +473,7 @@ compile_pip_requirements( py_library( name = "python_std", deps = [ + "//reboot/std/blobs/v1:blobs_py", "//reboot/std/ciphertext/v1:ciphertext_py", "//reboot/std/collections/ordered_map/v1:ordered_map_py", "//reboot/std/collections/queue/v1:queue_py", @@ -598,8 +599,12 @@ sh_library( ":reboot.dev", "//rbt/v1alpha1:reboot-dev-reboot-api", "//reboot/create-ui:reboot-dev-create-ui", + "//reboot/nodejs:reboot-dev-reboot-" + REBOOT_VERSION + ".tgz", "//reboot/react:reboot-dev-reboot-react", + "//reboot/std:reboot-dev-reboot-std", + "//reboot/std/react:reboot-dev-reboot-std-react", "//reboot/web:reboot-dev-reboot-web", + "@com_github_reboot_dev_reboot//rbt/std:reboot-dev-reboot-std-api", ], ) @@ -616,8 +621,12 @@ sh_binary( ":reboot.dev", "//rbt/v1alpha1:reboot-dev-reboot-api", "//reboot/create-ui:reboot-dev-create-ui", + "//reboot/nodejs:reboot-dev-reboot-" + REBOOT_VERSION + ".tgz", "//reboot/react:reboot-dev-reboot-react", + "//reboot/std:reboot-dev-reboot-std", + "//reboot/std/react:reboot-dev-reboot-std-react", "//reboot/web:reboot-dev-reboot-web", + "@com_github_reboot_dev_reboot//rbt/std:reboot-dev-reboot-std-api", ], deps = [":stage_and_publish_local_lib"], ) diff --git a/reboot/aio/BUILD.bazel b/reboot/aio/BUILD.bazel index 54ee332d..07307187 100644 --- a/reboot/aio/BUILD.bazel +++ b/reboot/aio/BUILD.bazel @@ -514,6 +514,7 @@ py_library( ":servicers_py", "//reboot/aio/auth:oauth_providers_py", "//reboot/aio/auth:oauth_server_py", + "//reboot/std/blobs/v1:blobs_py", ], ) diff --git a/reboot/aio/http.py b/reboot/aio/http.py index e4d91d2e..3c1ae418 100644 --- a/reboot/aio/http.py +++ b/reboot/aio/http.py @@ -149,6 +149,14 @@ def post(self, path: str, **kwargs): assert "methods" not in kwargs return self._api_route(path, methods=["POST"], **kwargs) + def put(self, path: str, **kwargs): + # Rather than list out all of the possible keyword args + # that `FastAPI` expects we'll just pass along any that + # are passed to us, but we don't expect `methods` as we + # override that below. + assert "methods" not in kwargs + return self._api_route(path, methods=["PUT"], **kwargs) + def options(self, path: str, **kwargs): # Used for CORS preflight handlers. assert "methods" not in kwargs diff --git a/reboot/aio/tests.py b/reboot/aio/tests.py index 67485fd6..73cfb717 100644 --- a/reboot/aio/tests.py +++ b/reboot/aio/tests.py @@ -2,6 +2,7 @@ import os import reboot.aio.reboot import secrets +import tempfile import unittest from reboot.aio.applications import Application, NodeApplication from reboot.aio.auth.oauth_providers import ( @@ -25,6 +26,8 @@ ENVVAR_REBOOT_ENABLE_EVENT_LOOP_BLOCKED_WATCHDOG, ENVVAR_REBOOT_IN_TEST, ) +from reboot.std.blobs.v1._data_plane import ENVVAR_BLOB_DATA_PLANE_URL +from reboot.std.blobs.v1._filesystem_server import FilesystemDataPlane from typing import ( Any, Awaitable, @@ -161,6 +164,47 @@ def __init__(self) -> None: os.environ[ENVVAR_REBOOT_IN_TEST] = 'true' # The application under test, or `None` before one is started. self._application: Optional[Application] = None + self._blob_data_plane: Optional[FilesystemDataPlane] = None + self._blob_data_plane_directory: Optional[tempfile.TemporaryDirectory + ] = None + + async def start(self): + result = await super().start() + # Run a filesystem blob data plane for the duration of the + # test, so that applications using `reboot.std.blobs` work in + # unit tests exactly as they do under `rbt dev run` (which + # spawns the same data plane). An already-configured data plane + # is honored, mirroring the CLI — including one set up by + # another live `Reboot` instance in this process, which then + # must outlive this instance's use of it. + if not os.environ.get(ENVVAR_BLOB_DATA_PLANE_URL): + self._blob_data_plane_directory = tempfile.TemporaryDirectory( + prefix="reboot-test-blobs-" + ) + self._blob_data_plane = await FilesystemDataPlane.start( + directory=self._blob_data_plane_directory.name, + ) + os.environ[ENVVAR_BLOB_DATA_PLANE_URL] = ( + self._blob_data_plane.url + ) + return result + + async def stop(self) -> None: + try: + await super().stop() + finally: + if self._blob_data_plane is not None: + await self._blob_data_plane.stop() + # Only clear the env var if it still points at our data + # plane; another `Reboot` instance may have replaced it + # with its own in the meantime. + if os.environ.get(ENVVAR_BLOB_DATA_PLANE_URL + ) == (self._blob_data_plane.url): + os.environ.pop(ENVVAR_BLOB_DATA_PLANE_URL, None) + self._blob_data_plane = None + if self._blob_data_plane_directory is not None: + self._blob_data_plane_directory.cleanup() + self._blob_data_plane_directory = None async def make_valid_oauth_access_token( self, @@ -414,6 +458,14 @@ async def up( # Should only have `application`, `local_envoy`, # `local_envoy_port`, `servers`, `effect_validation`. + # Do any pre-run library set up, just like `Application.run()` + # does; e.g. a library may register HTTP routes. Libraries must + # tolerate being `pre_run` more than once, since a test may + # `up` the same `Application` after a `down`. + if not in_nodejs(): + for library in application.libraries: + await library.pre_run(application) + # Check if application.http has methods or mounts (note this # isn't relevant for TypeScript, which doesn't have that # property). If yes, we need a local_envoy to be present to diff --git a/reboot/cli/commands/BUILD.bazel b/reboot/cli/commands/BUILD.bazel index 6f059898..066cb368 100644 --- a/reboot/cli/commands/BUILD.bazel +++ b/reboot/cli/commands/BUILD.bazel @@ -12,6 +12,7 @@ py_library( requirement("cryptography"), requirement("python-dotenv"), ":generate_py", + "//reboot/cli/common:blob_data_plane_py", "//reboot/cli/common:directories_py", "//reboot/cli/common:frontend_py", "//reboot/cli/common:monkeys_py", @@ -95,6 +96,7 @@ py_library( deps = [ requirement("aiofiles"), ":dev_py", + "//reboot/cli/common:blob_data_plane_py", "//reboot/cli/common:detect_cores_py", "//reboot/cli/common:directories_py", "//reboot/cli/common:frontend_py", diff --git a/reboot/cli/commands/dev.py b/reboot/cli/commands/dev.py index 00f10e6a..58affc3d 100644 --- a/reboot/cli/commands/dev.py +++ b/reboot/cli/commands/dev.py @@ -30,6 +30,10 @@ # We import the whole `terminal` module (as opposed to the methods it contains) # to allow us to mock these methods out in tests. from reboot.cli.common import terminal +from reboot.cli.common.blob_data_plane import ( + BLOBS_SUBDIRECTORY, + start_filesystem_data_plane, +) from reboot.cli.common.directories import ( add_working_directory_options, dot_rbt_dev_directory, @@ -80,6 +84,7 @@ RBT_APPLICATION_EXIT_CODE_BACKWARDS_INCOMPATIBILITY, LocalEnvoyMode, ) +from reboot.std.blobs.v1._data_plane import ENVVAR_BLOB_DATA_PLANE_URL from reboot.version import REBOOT_VERSION from typing import Any, Awaitable, Callable, Optional, TextIO, TypeVar @@ -1437,6 +1442,39 @@ def crypto_root_keys() -> str: root_keys_path.write_text(root_keys) return root_keys + # Honor a blob data plane configured via `--env-file`/`--env` (they + # are otherwise only composed into the application's environment at + # launch, below) so that no local one is spawned; mirror + # `compose_env`'s precedence: `--env` wins over `--env-file`. + if args.env_file is not None and os.path.isfile(args.env_file): + data_plane_url = _load_env_file(args.env_file + ).get(ENVVAR_BLOB_DATA_PLANE_URL) + if data_plane_url is not None: + env[ENVVAR_BLOB_DATA_PLANE_URL] = data_plane_url + for (key, value) in args.env or []: + if key == ENVVAR_BLOB_DATA_PLANE_URL: + env[key] = value + + # Start the filesystem blob data plane (unless a data plane is + # already configured) and point the application at it. Started + # with `env` — after the cryptographic root keys above, which its + # URL signing derives from. Bytes live under the application's + # state directory so `rbt dev expunge` removes them along with the + # rest of the state. + blobs_directory = os.path.join( + env.get( + ENVVAR_RBT_STATE_DIRECTORY, + str(dot_rbt_dev_directory(args, parser)), + ), + BLOBS_SUBDIRECTORY, + ) + await start_filesystem_data_plane( + env, + blobs_directory, + subprocesses, + background_command_tasks, + ) + if tracing == Tracing.JAEGER: # TODO: dynamic port. See comment in `_run_jaeger()`. env[OTEL_EXPORTER_OTLP_TRACES_ENDPOINT] = "localhost:4317" diff --git a/reboot/cli/commands/serve.py b/reboot/cli/commands/serve.py index c1ac74f1..58a58ad9 100644 --- a/reboot/cli/commands/serve.py +++ b/reboot/cli/commands/serve.py @@ -1,5 +1,6 @@ import aiofiles.os import argparse +import asyncio import math import os import sys @@ -13,6 +14,10 @@ try_and_become_child_subreaper_on_linux, ) from reboot.cli.common import terminal +from reboot.cli.common.blob_data_plane import ( + BLOBS_SUBDIRECTORY, + start_filesystem_data_plane, +) from reboot.cli.common.detect_cores import detect_cores from reboot.cli.common.directories import ( add_working_directory_options, @@ -325,6 +330,24 @@ async def serve_run( for (key, value) in args.env or []: env[key] = value + # Start the filesystem blob data plane (unless a data plane is + # already configured, e.g. by Reboot Cloud provisioning or via + # `--env`) and point the application at it. Bytes live under the + # state directory alongside the rest of the application's state. + # Done after `--env` is applied so an explicitly-configured data + # plane is honored rather than spawning a redundant local one. + blobs_directory = os.path.join( + env.get(ENVVAR_RBT_STATE_DIRECTORY, os.getcwd()), + BLOBS_SUBDIRECTORY, + ) + data_plane_tasks: list[asyncio.Task] = [] + await start_filesystem_data_plane( + env, + blobs_directory, + subprocesses, + data_plane_tasks, + ) + # If 'PYTHONPATH' is not explicitly set, we'll set it to the # specified generated code directory plus each proto directory. # We want to get the directories from 'rbt generate' flags, @@ -381,8 +404,16 @@ async def serve_run( application if not auto_transpilation else str(bundle), ] - async with subprocesses.exec(*args, env=env) as process: - return await process.wait() + try: + async with subprocesses.exec(*args, env=env) as process: + return await process.wait() + finally: + for data_plane_task in data_plane_tasks: + data_plane_task.cancel() + try: + await data_plane_task + except asyncio.CancelledError: + pass async def handle_serve_subcommand( diff --git a/reboot/cli/common/BUILD.bazel b/reboot/cli/common/BUILD.bazel index de53cec0..1573c76c 100644 --- a/reboot/cli/common/BUILD.bazel +++ b/reboot/cli/common/BUILD.bazel @@ -93,6 +93,17 @@ py_library( ], ) +py_library( + name = "blob_data_plane_py", + srcs = ["blob_data_plane.py"], + srcs_version = "PY3", + visibility = ["//visibility:public"], + deps = [ + ":subprocesses_py", + "@com_github_reboot_dev_reboot//reboot/std/blobs/v1:blobs_py", + ], +) + py_library( name = "monkeys_py", srcs = ["monkeys.py"], diff --git a/reboot/cli/common/blob_data_plane.py b/reboot/cli/common/blob_data_plane.py new file mode 100644 index 00000000..23c512dc --- /dev/null +++ b/reboot/cli/common/blob_data_plane.py @@ -0,0 +1,141 @@ +"""Spawning the filesystem blob data plane for local runs. + +`rbt dev run` and `rbt serve run` start the open-source filesystem blob +data-plane server (`reboot.std.blobs.v1._filesystem_server`) as a +background subprocess whenever `REBOOT_BLOB_DATA_PLANE_URL` is not +already set, and point the application at it on localhost. In Reboot +Cloud the variable is set by provisioning (to the app's facilitator), +so nothing is spawned. The server is started unconditionally when the +variable is unset — it is cheap and idle unless the application +actually uses blobs. + +The server chooses its own ports and writes them to a "ready file" +once both its gRPC and HTTP endpoints are listening; we wait for that +file before pointing the application at it. That both avoids the +port-allocation race of picking a port in the parent and guarantees +the data plane is fully serving before the application can reach it. +""" + +import asyncio +import os +import shutil +import sys +import tempfile +from reboot.cli.common.subprocesses import Subprocesses +from reboot.std.blobs.v1._data_plane import ENVVAR_BLOB_DATA_PLANE_URL +from typing import Optional + +# The filesystem data plane binds loopback only (its gRPC control +# surface is unauthenticated); the application reaches it here. +_LOOPBACK_HOST = "127.0.0.1" + +# Where blob bytes live, under the application's state directory (so +# `rbt dev expunge`, which removes the state directory, removes blobs +# too). +BLOBS_SUBDIRECTORY = "blobs" + +# How long to wait for the spawned server to report its ports before +# giving up. +_READY_TIMEOUT_SECONDS = 30 +_READY_POLL_INTERVAL_SECONDS = 0.1 + + +async def _run_and_await_ready( + subprocesses: Subprocesses, + argv: list[str], + env: dict[str, str], + ready_file: str, + ready: asyncio.Future[tuple[int, int]], +) -> None: + """Runs the data-plane server until cancelled, resolving `ready` + with its `(grpc_port, http_port)` once the server writes them.""" + async with subprocesses.exec(*argv, env=env) as process: + waited = 0.0 + while not ready.done(): + if process.returncode is not None: + ready.set_exception( + RuntimeError( + "The filesystem blob data-plane server exited " + f"before becoming ready (code {process.returncode})." + ) + ) + break + ports = _read_ready_file(ready_file) + if ports is not None: + ready.set_result(ports) + break + if waited >= _READY_TIMEOUT_SECONDS: + ready.set_exception( + RuntimeError( + "Timed out waiting for the filesystem blob " + "data-plane server to become ready." + ) + ) + break + await asyncio.sleep(_READY_POLL_INTERVAL_SECONDS) + waited += _READY_POLL_INTERVAL_SECONDS + await process.wait() + + +def _read_ready_file(path: str) -> Optional[tuple[int, int]]: + try: + with open(path) as f: + content = f.read().split() + except FileNotFoundError: + return None + if len(content) != 2: + return None + return int(content[0]), int(content[1]) + + +async def start_filesystem_data_plane( + env: dict[str, str], + blobs_directory: str, + subprocesses: Subprocesses, + background_command_tasks: list[asyncio.Task], +) -> None: + """Unless a data plane is already configured in `env`, spawns the + filesystem server as a background task, waits until it is fully + listening, and points `env[REBOOT_BLOB_DATA_PLANE_URL]` at it. The + server runs with `env` — which must already carry + `REBOOT_CRYPTO_ROOT_KEYS`, from which the server derives its + URL-signing key — and is cleaned up when its task is cancelled.""" + if env.get(ENVVAR_BLOB_DATA_PLANE_URL): + return + + # The ready file lives in a fresh private directory (a bare + # `mktemp` name in a shared `/tmp` would be squattable). + ready_directory = tempfile.mkdtemp(prefix="reboot-blob-data-plane-") + ready_file = os.path.join(ready_directory, "ready") + argv = [ + sys.executable, + "-m", + "reboot.std.blobs.v1._filesystem_server", + "--directory", + blobs_directory, + "--ready-file", + ready_file, + ] + + loop = asyncio.get_event_loop() + ready: asyncio.Future[tuple[int, int]] = loop.create_future() + task = asyncio.create_task( + _run_and_await_ready(subprocesses, argv, env, ready_file, ready), + name="run_filesystem_data_plane(...)", + ) + background_command_tasks.append(task) + try: + grpc_port, _ = await ready + except BaseException: + # Reap the never-ready server immediately: its task would + # otherwise run until the caller's cleanup — which under + # `rbt serve run` only happens once the application exits. + task.cancel() + try: + await task + except BaseException: + pass + raise + finally: + shutil.rmtree(ready_directory, ignore_errors=True) + env[ENVVAR_BLOB_DATA_PLANE_URL] = f"{_LOOPBACK_HOST}:{grpc_port}" diff --git a/reboot/examples/chat-room/api/chat_room/v1/chat_room.proto b/reboot/examples/chat-room/api/chat_room/v1/chat_room.proto index f3c43780..f21d1875 100644 --- a/reboot/examples/chat-room/api/chat_room/v1/chat_room.proto +++ b/reboot/examples/chat-room/api/chat_room/v1/chat_room.proto @@ -6,10 +6,29 @@ import "rbt/v1alpha1/options.proto"; //////////////////////////////////////////////////////////////////////// +// Raised when a requested attachment exceeds the chat room's +// per-attachment size limit. +message AttachmentTooLarge { + // The maximum allowed attachment size, in bytes, so the client can + // show a helpful message. + uint64 max_size_bytes = 1; +} + +message Message { + string text = 1; + + // Ids of `rbt.std.blobs.v1.Blob`s holding this message's + // attachments. Attachments are visible to all participants from the + // moment the message is sent, while their bytes may still be + // uploading; render their progress reactively via `Blob.Info`. + repeated string attachment_blob_ids = 2; +} + message ChatRoom { option (rbt.v1alpha1.state) = { }; - repeated string messages = 1; + reserved 1; + repeated Message messages = 2; } service ChatRoomMethods { @@ -19,9 +38,14 @@ service ChatRoomMethods { }; } - // Adds a new message to the list of recorded messages. + // Adds a new message to the list of recorded messages, creating a + // `Blob` for each requested attachment. The message is published + // immediately; the caller then uploads the attachment bytes into + // the returned blob ids. rpc Send(SendRequest) returns (SendResponse) { - option (rbt.v1alpha1.method).writer = { + option (rbt.v1alpha1.method) = { + transaction: {}, + errors: [ "AttachmentTooLarge" ], }; } } @@ -29,11 +53,23 @@ service ChatRoomMethods { message MessagesRequest {} message MessagesResponse { - repeated string messages = 1; + repeated Message messages = 1; +} + +// Describes an attachment the sender intends to upload. +message Attachment { + string content_type = 1; // E.g. "image/png". + uint64 size_bytes = 2; } message SendRequest { string message = 1; // E.g. "Hello, World". + repeated Attachment attachments = 2; } -message SendResponse {} +message SendResponse { + // One `Blob` id per requested attachment, in request order. Upload + // the attachment bytes into these (e.g. with `useBlobUpload` from + // `@reboot-dev/reboot-std-react/blobs`). + repeated string attachment_blob_ids = 1; +} diff --git a/reboot/examples/chat-room/backend/src/chat_room_servicer.py b/reboot/examples/chat-room/backend/src/chat_room_servicer.py index eaa0adc9..48eb6d40 100644 --- a/reboot/examples/chat-room/backend/src/chat_room_servicer.py +++ b/reboot/examples/chat-room/backend/src/chat_room_servicer.py @@ -1,12 +1,18 @@ from chat_room.v1.chat_room_rbt import ( + AttachmentTooLarge, ChatRoom, + Message, MessagesRequest, MessagesResponse, SendRequest, SendResponse, ) +from rbt.std.blobs.v1.blobs_rbt import Blob from reboot.aio.auth.authorizers import allow -from reboot.aio.contexts import ReaderContext, WriterContext +from reboot.aio.contexts import ReaderContext, TransactionContext + +# The chat room refuses attachments larger than this. +MAX_ATTACHMENT_BYTES = 300 * 1024 * 1024 class ChatRoomServicer(ChatRoom.Servicer): @@ -23,9 +29,37 @@ async def messages( async def send( self, - context: WriterContext, + context: TransactionContext, request: SendRequest, ) -> SendResponse: - message = request.message - self.state.messages.extend([message]) - return SendResponse() + # Create a `Blob` for every requested attachment. This is the + # application-mediated step where attachment policy is + # enforced; the blobs' random ids then act as upload/download + # capabilities. This example has no end-user authentication, + # so `uploader_id` is left empty: anyone who knows a blob's id + # may upload into it. `downloader_ids` is likewise omitted, so + # anyone who knows a blob's id may download it too. + attachment_blob_ids = [] + for attachment in request.attachments: + if attachment.size_bytes > MAX_ATTACHMENT_BYTES: + raise ChatRoom.SendAborted( + AttachmentTooLarge(max_size_bytes=MAX_ATTACHMENT_BYTES) + ) + blob, _ = await Blob.create( + context, + content_type=attachment.content_type, + size=attachment.size_bytes, + max_size=MAX_ATTACHMENT_BYTES, + ) + attachment_blob_ids.append(blob.state_id) + + # The message is published immediately: participants see it + # (and can watch its attachments' upload progress) before the + # bytes have been uploaded. + self.state.messages.append( + Message( + text=request.message, + attachment_blob_ids=attachment_blob_ids, + ) + ) + return SendResponse(attachment_blob_ids=attachment_blob_ids) diff --git a/reboot/examples/chat-room/backend/src/main.py b/reboot/examples/chat-room/backend/src/main.py index 9683a229..dc840752 100644 --- a/reboot/examples/chat-room/backend/src/main.py +++ b/reboot/examples/chat-room/backend/src/main.py @@ -4,6 +4,7 @@ from chat_room_servicer import ChatRoomServicer from reboot.aio.applications import Application from reboot.aio.external import InitializeContext +from reboot.std.blobs.v1.blobs import blobs_library logging.basicConfig(level=logging.INFO) @@ -25,6 +26,10 @@ async def initialize(context: InitializeContext): async def main(): await Application( servicers=[ChatRoomServicer], + # Message attachments are stored as blobs; `rbt dev run` and + # `rbt serve run` provide a local filesystem data plane (see + # `REBOOT_BLOB_DATA_PLANE_URL` for using a custom one). + libraries=[blobs_library()], initialize=initialize, ).run() diff --git a/reboot/examples/chat-room/frontend/.tests/type_check.sh b/reboot/examples/chat-room/frontend/.tests/type_check.sh index ea751190..a58789ed 100755 --- a/reboot/examples/chat-room/frontend/.tests/type_check.sh +++ b/reboot/examples/chat-room/frontend/.tests/type_check.sh @@ -60,7 +60,9 @@ if [[ -n "${REBOOT_NPM_PACKAGE:-}" ]]; then "${SANDBOX_ROOT}${REBOOT_NPM_PACKAGE}" \ "${SANDBOX_ROOT}${REBOOT_API_NPM_PACKAGE}" \ "${SANDBOX_ROOT}${REBOOT_WEB_NPM_PACKAGE}" \ - "${SANDBOX_ROOT}${REBOOT_REACT_NPM_PACKAGE}" + "${SANDBOX_ROOT}${REBOOT_REACT_NPM_PACKAGE}" \ + "${SANDBOX_ROOT}${REBOOT_STD_API_PACKAGE}" \ + "${SANDBOX_ROOT}${REBOOT_STD_REACT_PACKAGE}" else npm install fi diff --git a/reboot/examples/chat-room/frontend/reboot-non-react-web/src/main.ts b/reboot/examples/chat-room/frontend/reboot-non-react-web/src/main.ts index 97e7cffd..666df24c 100644 --- a/reboot/examples/chat-room/frontend/reboot-non-react-web/src/main.ts +++ b/reboot/examples/chat-room/frontend/reboot-non-react-web/src/main.ts @@ -33,7 +33,7 @@ async function bindToElement( ) { for await (const response of generator) { element.innerHTML = `${response.messages - .map((msg: string) => `
${msg}
`) + .map((msg) => `
${msg.text}
`) .join("")}`; } } diff --git a/reboot/examples/chat-room/frontend/web/package.json b/reboot/examples/chat-room/frontend/web/package.json index 4fee9993..425d69f4 100644 --- a/reboot/examples/chat-room/frontend/web/package.json +++ b/reboot/examples/chat-room/frontend/web/package.json @@ -7,6 +7,8 @@ "@bufbuild/protobuf": "1.10.1", "@eslint/js": "^9.34.0", "@reboot-dev/reboot-react": "1.3.0", + "@reboot-dev/reboot-std-api": "1.3.0", + "@reboot-dev/reboot-std-react": "1.3.0", "@types/eslint__js": "^8.42.3", "@types/react": "^19.1.0", "@types/react-dom": "^19.1.0", diff --git a/reboot/examples/chat-room/frontend/web/src/App.module.css b/reboot/examples/chat-room/frontend/web/src/App.module.css index 2289ef53..24744a8a 100644 --- a/reboot/examples/chat-room/frontend/web/src/App.module.css +++ b/reboot/examples/chat-room/frontend/web/src/App.module.css @@ -1,22 +1,127 @@ -/* Styles by @sowg: https://codepen.io/sowg/pen/qBXexaZ */ +/* Styles originally by @sowg: https://codepen.io/sowg/pen/qBXexaZ */ .messages { - width: 300px; + width: 340px; margin: 0 auto; } +/* --- Composer: a rounded text field with a `+` attach button inside, + and a Send button beside it. --- */ + +.composer { + display: flex; + align-items: center; + gap: 0.5em; + margin: 1em 0 0.4em; +} + +.inputWrap { + position: relative; + flex: 1; +} + .textInput { - margin: 1em; + width: 100%; + box-sizing: border-box; appearance: none; - border: none; + border: 0.15em solid #e91e63; outline: none; - border-bottom: 0.2em solid #e91e63; - background: rgba(#e91e63, 0.2); - border-radius: 0.2em 0.2em 0 0; - padding: 0.4em; + border-radius: 1.6em; + padding: 0.55em 2.5em 0.55em 0.9em; + color: #e91e63; + background: #fff0f5; + font-family: sans-serif; +} + +.textInput::placeholder { + color: #f48fb1; +} + +.plusButton { + position: absolute; + /* Equal inset on the top, bottom, and right; `aspect-ratio` then + makes it a circle sized to fit, so its right margin matches its + vertical margin. */ + top: 0.35em; + bottom: 0.35em; + right: 0.35em; + aspect-ratio: 1 / 1; + padding: 0; + border: none; + border-radius: 50%; + background: #e91e63; + color: #fff; + font-size: 1em; + line-height: 1; + cursor: pointer; + display: flex; + align-items: center; + justify-content: center; +} + +.plusButton:hover { + background: #c2185b; +} + +.hiddenFileInput { + display: none; +} + +/* A dismissable error notification (e.g. an attachment that's too + large). */ +.errorBanner { + display: flex; + align-items: center; + justify-content: space-between; + gap: 0.6em; + margin: 0 0 0.6em; + padding: 0.6em 0.9em; + border-radius: 0.8em; + background: #ffe3ec; + border: 0.12em solid #e91e63; + color: #c2185b; + font-family: sans-serif; + font-size: 0.9em; + cursor: pointer; +} + +.errorDismiss { + font-size: 1.2em; + line-height: 1; + flex: none; +} + +/* A chip showing the file that's staged to send, before the message + is sent. */ +.pendingAttachment { + display: flex; + align-items: center; + gap: 0.3em; + margin: 0 0 0.6em 0.4em; + font-family: sans-serif; + font-size: 0.85em; + color: #e91e63; +} + +.pendingAttachmentName { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; + max-width: 220px; +} + +.pendingAttachmentRemove { + border: none; + background: transparent; color: #e91e63; + font-size: 1.2em; + line-height: 1; + cursor: pointer; + padding: 0 0.2em; } +/* --- Messages --- */ + .message { appearance: none; border: 0.2em solid #e91e63; @@ -29,6 +134,10 @@ font-family: sans-serif; } +.messageText { + word-break: break-word; +} + .pendingMessage { border: 0.2em solid #a9a9a9; color: #a9a9a9; @@ -47,18 +156,21 @@ .buttonDisabled, .buttonEnabled { - float: right; + flex: none; appearance: none; - border: 0.2em solid; + border: 0.15em solid; background: hsl(0 0 0/0); - padding: 0.85em 1.5em; + padding: 0.55em 1.1em; border-radius: 2em; - transition: 1s; + font-family: sans-serif; + transition: 0.3s; + cursor: pointer; } .buttonDisabled { color: gray; border-color: darkgray; + cursor: default; } .buttonEnabled { @@ -71,3 +183,94 @@ color: #fff; } } + +/* --- Attachments: fixed clickable squares --- */ + +.attachments { + display: flex; + flex-wrap: wrap; + gap: 0.5em; + margin-top: 0.6em; +} + +.attachment, +.attachmentPending { + width: 72px; + height: 72px; + flex: none; + box-sizing: border-box; + padding: 0; + border: 0.15em solid #e91e63; + border-radius: 0.6em; + background: #fff0f5; + overflow: hidden; + display: flex; + align-items: center; + justify-content: center; +} + +.attachment { + cursor: pointer; +} + +.attachment:disabled { + cursor: default; +} + +.attachmentPending { + border-color: #cfcfcf; + background: #f4f4f4; +} + +.attachmentThumb { + width: 100%; + height: 100%; + object-fit: cover; + display: block; +} + +.attachmentGlyph { + font-size: 1.9em; + color: #a9a9a9; +} + +/* Circular upload progress. Rotated so it fills from the top. */ +.ring { + width: 40px; + height: 40px; + transform: rotate(-90deg); +} + +.ringTrack { + fill: none; + stroke: #f8bbd0; + stroke-width: 3; +} + +.ringFill { + fill: none; + stroke: #e91e63; + stroke-width: 3; + stroke-linecap: round; + transition: stroke-dashoffset 0.3s ease; +} + +/* --- Lightbox for viewing a committed image attachment --- */ + +.lightbox { + position: fixed; + inset: 0; + z-index: 10; + background: rgba(0, 0, 0, 0.75); + display: flex; + align-items: center; + justify-content: center; + cursor: zoom-out; +} + +.lightboxImage { + max-width: 90vw; + max-height: 90vh; + border-radius: 0.4em; + box-shadow: 0 0.5em 2em rgba(0, 0, 0, 0.5); +} diff --git a/reboot/examples/chat-room/frontend/web/src/App.tsx b/reboot/examples/chat-room/frontend/web/src/App.tsx index 123ea889..526201ea 100644 --- a/reboot/examples/chat-room/frontend/web/src/App.tsx +++ b/reboot/examples/chat-room/frontend/web/src/App.tsx @@ -1,18 +1,129 @@ -import { FC, useState } from "react"; +import { Blob_Status } from "@reboot-dev/reboot-std-api/blobs/v1/blobs_pb.js"; +import { + useBlob, + useBlobDownloadUrl, + useBlobUpload, +} from "@reboot-dev/reboot-std-react/blobs"; +import { FC, useRef, useState } from "react"; import css from "./App.module.css"; -import { useChatRoom } from "../../api/chat_room/v1/chat_room_rbt_react"; +import { + AttachmentTooLarge, + useChatRoom, +} from "../../api/chat_room/v1/chat_room_rbt_react"; // We can choose any id we want because the state will be constructed when we // make the first .writer call. const STATE_MACHINE_ID = "reboot-chat-room"; -const Message: FC<{ text: string }> = ({ text }) => { - return
{text}
; +// A circular upload-progress indicator that fills clockwise. `fraction` +// is 0..1; when the total size isn't known yet it stays near empty. +const ProgressRing: FC<{ fraction: number }> = ({ fraction }) => { + const radius = 13; + const circumference = 2 * Math.PI * radius; + return ( + + + + + ); +}; + +// A fixed-size square for one attachment. While the blob is uploading +// it shows a filling ring (progress lives on the blob's state, so this +// is visible to every participant); once committed it shows a +// thumbnail (or a file glyph) that can be clicked to view the content. +const Attachment: FC<{ blobId: string }> = ({ blobId }) => { + const { response: info } = useBlob({ id: blobId }).useInfo(); + const url = useBlobDownloadUrl(blobId); + const [viewing, setViewing] = useState(false); + + const isImage = info?.contentType.startsWith("image/") ?? false; + const committed = info?.status === Blob_Status.COMMITTED && url !== undefined; + const gone = + info?.status === Blob_Status.DELETING || + info?.status === Blob_Status.DELETED; + + const total = Number(info?.size ?? 0); + const uploaded = Number(info?.bytesUploaded ?? 0); + const fraction = total > 0 ? uploaded / total : 0; + + const view = () => { + if (!committed) return; + if (isImage) { + setViewing(true); + } else { + window.open(url, "_blank", "noopener"); + } + }; + + return ( + <> + + {viewing && committed && isImage && ( +
setViewing(false)}> + attachment +
+ )} + + ); }; -const PendingMessage: FC<{ text: string; isLoading: boolean }> = ({ +const Attachments: FC<{ blobIds: string[] }> = ({ blobIds }) => + blobIds.length > 0 ? ( +
+ {blobIds.map((blobId) => ( + + ))} +
+ ) : null; + +const Message: FC<{ text: string; attachmentBlobIds: string[] }> = ({ text, - isLoading, + attachmentBlobIds, }) => { + return ( +
+ {text !== "" &&
{text}
} + +
+ ); +}; + +// An optimistically-rendered message that hasn't been confirmed by the +// backend yet. Its attachments have no blob id, so we show empty +// placeholder squares to keep the layout stable until the real message +// arrives (with squares that then track upload progress). +const PendingMessage: FC<{ + text: string; + attachmentCount: number; + isLoading: boolean; +}> = ({ text, attachmentCount, isLoading }) => { return (
= ({ : `${css.message} ${css.pendingMessage}` } > - {text} + {text !== "" &&
{text}
} + {attachmentCount > 0 && ( +
+ {Array.from({ length: attachmentCount }, (_, index) => ( +
+ +
+ ))} +
+ )}
); }; const App = () => { - // State of the input component. + // State of the input components. const [message, setMessage] = useState("Hello, Reboot!"); + const [file, setFile] = useState(null); + const [error, setError] = useState(null); + const fileInputRef = useRef(null); const { useMessages, send } = useChatRoom({ id: STATE_MACHINE_ID }); const { response } = useMessages(); + const { upload } = useBlobUpload(); - const handleClick = async () => { - const { aborted } = await send({ message: message }); + const clearFile = () => { + setFile(null); + if (fileInputRef.current !== null) { + fileInputRef.current.value = ""; + } + }; + + const canSend = message !== "" || file !== null; + + const handleSend = async () => { + if (!canSend) { + return; + } + setError(null); + // Publishes the message immediately; the backend creates a `Blob` + // per requested attachment and returns its id. Every participant + // sees the message right away, with attachments visibly uploading. + const { response, aborted } = await send({ + message: message, + attachments: + file !== null + ? [{ contentType: file.type, sizeBytes: BigInt(file.size) }] + : [], + }); if (aborted !== undefined) { - console.warn(aborted.error.getType()); - console.warn(aborted.message); + // Surface the failure to the user. The backend's + // `AttachmentTooLarge` error carries the limit, so we can say + // exactly what it is. + if (aborted.error instanceof AttachmentTooLarge) { + const maxMebibytes = Number(aborted.error.maxSizeBytes) / (1024 * 1024); + setError( + `That attachment is too large. The maximum is ${maxMebibytes} MiB.` + ); + } else { + setError(`Couldn't send your message: ${aborted.message}`); + } + return; } + const uploadFile = file; setMessage(""); + clearFile(); + + if (uploadFile !== null && response !== undefined) { + // Uploads directly to the data plane (the app's filesystem + // store locally, S3 on the Cloud -- this code neither knows + // nor cares), reporting progress onto the blob's state as it + // goes. + const { error } = await upload(response.attachmentBlobIds[0], uploadFile); + if (error !== undefined) { + console.warn(`Attachment upload failed: ${error}`); + } + } }; return (
+
+
+ + setMessage(e.target.value)} + onKeyDown={(e) => { + if (e.key === "Enter" && canSend) { + handleSend(); + } + }} + value={message} + placeholder="Your message here..." + /> +
+ +
+ setMessage(e.target.value)} - onKeyDown={(e) => { - if (e.key === "Enter" && message !== "") { - handleClick(); - } + ref={fileInputRef} + type="file" + className={css.hiddenFileInput} + onChange={(e) => { + setError(null); + setFile(e.target.files?.[0] ?? null); }} - value={message} - placeholder="Your message here..." /> - + + {error !== null && ( +
setError(null)} + title="Dismiss" + > + {error} + × +
+ )} + + {file !== null && ( +
+ 📎 {file.name} + +
+ )} + {(response !== undefined && response.messages.length > 0 && - response.messages.map((message: string) => ( - + response.messages.map((message, index) => ( + ))) || (response !== undefined && response.messages.length === 0 && (

No messages yet!

@@ -78,8 +299,13 @@ const App = () => { been received that includes the mutation's updates so you don't have to worry about mutators racing with readers! */} - {send.pending.map(({ request: { message }, isLoading }) => ( - + {send.pending.map(({ request, isLoading }, index) => ( + ))} {/* If we're loading our first response, show the user a loading message, diff --git a/reboot/routing/cors_settings.py b/reboot/routing/cors_settings.py index 3bb56494..a100fdfb 100644 --- a/reboot/routing/cors_settings.py +++ b/reboot/routing/cors_settings.py @@ -39,8 +39,10 @@ 'ngrok-skip-browser-warning', ) -# Response headers cross-origin JavaScript is allowed to read. -CORS_EXPOSE_HEADERS = ('grpc-status', 'grpc-message') +# Response headers cross-origin JavaScript is allowed to read. `etag` +# is exposed so that browser blob uploads can read each part's ETag +# from the data plane's response (see `reboot.std.blobs`). +CORS_EXPOSE_HEADERS = ('grpc-status', 'grpc-message', 'etag') # How long a browser may cache a CORS preflight response, in seconds # (20 days). diff --git a/reboot/std/BUILD.bazel b/reboot/std/BUILD.bazel index 8c81b120..e33e1fc8 100644 --- a/reboot/std/BUILD.bazel +++ b/reboot/std/BUILD.bazel @@ -47,6 +47,7 @@ ts_project( }, visibility = ["//visibility:public"], deps = [ + "//reboot/std/blobs/v1:blobs_ts", "//reboot/std/ciphertext/v1:ciphertext_ts", "//reboot/std/collections/ordered_map/v1:ordered_map_ts", "//reboot/std/collections/queue/v1:queue_ts", diff --git a/reboot/std/blobs/v1/BUILD.bazel b/reboot/std/blobs/v1/BUILD.bazel new file mode 100644 index 00000000..1fc2bb39 --- /dev/null +++ b/reboot/std/blobs/v1/BUILD.bazel @@ -0,0 +1,54 @@ +load("@aspect_rules_ts//ts:defs.bzl", "ts_project") +load("@rbt_pypi//:requirements.bzl", "requirement") +load("@rules_python//python:defs.bzl", "py_library") + +py_library( + name = "blobs_py", + srcs = [ + "_data_plane.py", + "_filesystem_server.py", + "_http.py", + "_proxy.py", + "_store.py", + "blobs.py", + ], + visibility = ["//visibility:public"], + deps = [ + "//log:log_py", + "//reboot/aio:applications_py", + "//reboot/aio:contexts_py", + "//reboot/aio:http_py", + "//reboot/aio:workflows_py", + "//reboot/aio/auth:authorizers_py", + "//reboot/crypto:root_keys_py", + "@com_github_reboot_dev_reboot//rbt/std/blobs/v1:blobs_py_reboot", + "@com_github_reboot_dev_reboot//rbt/std/blobs/v1:data_plane_py_reboot", + requirement("aiohttp"), + requirement("grpcio"), + requirement("starlette"), + requirement("uvicorn"), + ], +) + +ts_project( + name = "blobs_ts", + srcs = [ + "index.ts", + "package.json", + ], + declaration = True, + tsconfig = { + "compilerOptions": { + "declaration": True, + "module": "nodenext", + "moduleResolution": "nodenext", + "target": "es2020", + }, + }, + visibility = ["//visibility:public"], + deps = [ + "//:node_modules/@reboot-dev/reboot", + "//:node_modules/@reboot-dev/reboot-std-api", + "//:node_modules/@types/node", + ], +) diff --git a/reboot/std/blobs/v1/_data_plane.py b/reboot/std/blobs/v1/_data_plane.py new file mode 100644 index 00000000..d7f97cdc --- /dev/null +++ b/reboot/std/blobs/v1/_data_plane.py @@ -0,0 +1,93 @@ +"""Client-side glue for talking to a `BlobDataPlane` gRPC service. + +The `Blob` control-plane servicer holds no bytes; it calls a data-plane +service (discovered via `REBOOT_BLOB_DATA_PLANE_URL`) to provision +uploads, finalize them, mint URLs, and delete objects. Every local run +mode provides a filesystem data plane automatically: `rbt dev run` and +`rbt serve run` spawn it as a subprocess, and the +`reboot.aio.tests.Reboot` harness runs it in-process. Deployments set +the URL themselves (Reboot Cloud points it at the app's facilitator). +""" + +import grpc +import os +from rbt.std.blobs.v1.data_plane_pb2_grpc import BlobDataPlaneStub +from urllib.parse import urlparse + +# The URL of the `BlobDataPlane` gRPC service. A bare `host:port`, or a +# URL whose scheme selects transport security (`https`/`grpcs` -> +# secure, anything else -> insecure). Set by `rbt dev run`/`rbt serve +# run` and the test harness (which provide the filesystem data plane +# when it is unset) or by Reboot Cloud provisioning (pointing at the +# app's facilitator). +ENVVAR_BLOB_DATA_PLANE_URL = "REBOOT_BLOB_DATA_PLANE_URL" + +# The path namespace a data plane's forwarded paths must live under; +# the application refuses to forward anything else, so a data plane +# can never claim application routes (see `ForwardedPath` in +# `data_plane.proto`). +FORWARDED_PATH_PREFIX = "/__/reboot/blob/" + +_SECURE_SCHEMES = ("https", "grpcs") + + +class DataPlaneNotConfigured(RuntimeError): + """Raised when no data-plane URL is configured. Under `rbt dev + run`/`rbt serve run` and in `reboot.aio.tests.Reboot` unit tests + this never happens (they provide the filesystem data plane and set + the URL); it indicates the application was started some other way + without a data plane.""" + + +def channel_for_url(url: str) -> grpc.aio.Channel: + """Builds a gRPC channel to the data-plane service at `url`. The + scheme selects transport security; the host:port is the gRPC + target. Any URL path is ignored (gRPC addresses by host:port, not + path).""" + # `urlparse` needs a scheme to populate `netloc`; treat a bare + # `host:port` as such. + parsed = urlparse(url if "://" in url else f"grpc://{url}") + target = parsed.netloc + if parsed.scheme in _SECURE_SCHEMES: + return grpc.aio.secure_channel(target, grpc.ssl_channel_credentials()) + return grpc.aio.insecure_channel(target) + + +def proxy_target_from_environment(http_port: int) -> str: + """The base URL of the data plane's HTTP byte endpoint, to which + the application proxies forwarded paths: the host the application + already reaches the data plane's gRPC service on (from + `REBOOT_BLOB_DATA_PLANE_URL`), with the `http_port` its + `Configuration` reported, over the same transport security.""" + url = os.environ.get(ENVVAR_BLOB_DATA_PLANE_URL) + if not url: + raise DataPlaneNotConfigured( + f"`{ENVVAR_BLOB_DATA_PLANE_URL}` is not set." + ) + parsed = urlparse(url if "://" in url else f"grpc://{url}") + scheme = "https" if parsed.scheme in _SECURE_SCHEMES else "http" + host = parsed.hostname or "" + # `urlparse` strips the brackets off an IPv6 literal; put them + # back, since they are required in a URL authority. + if ":" in host: + host = f"[{host}]" + return f"{scheme}://{host}:{http_port}" + + +def stub_from_environment() -> BlobDataPlaneStub: + """Builds a data-plane stub from `REBOOT_BLOB_DATA_PLANE_URL`. A + fresh channel is created on the current event loop (rather than + memoized) because `grpc.aio` channels are event-loop-affine, and + an application may be brought up on a new loop. Called once per + library `pre_run`; the channel then lives as long as the + application (the `Library` has no shutdown hook on which to close + it).""" + url = os.environ.get(ENVVAR_BLOB_DATA_PLANE_URL) + if not url: + raise DataPlaneNotConfigured( + f"`{ENVVAR_BLOB_DATA_PLANE_URL}` is not set. Run the " + "application via `rbt dev run` or `rbt serve run` (which " + "start the filesystem blob data plane automatically), or " + "set the variable to your own data-plane service's URL." + ) + return BlobDataPlaneStub(channel_for_url(url)) diff --git a/reboot/std/blobs/v1/_filesystem_server.py b/reboot/std/blobs/v1/_filesystem_server.py new file mode 100644 index 00000000..6ec32aa6 --- /dev/null +++ b/reboot/std/blobs/v1/_filesystem_server.py @@ -0,0 +1,318 @@ +"""The filesystem blob data-plane server. + +Implements the `BlobDataPlane` gRPC service backed by the local +filesystem, plus the HTTP byte endpoint its minted URLs point at. It +runs in two ways: `rbt dev run` and `rbt serve run` spawn it as a +standalone program (via `main`) whenever `REBOOT_BLOB_DATA_PLANE_URL` +is not already set, and the `reboot.aio.tests.Reboot` test harness +runs it in-process (via `FilesystemDataPlane.start`), so that blob +storage works out of the box in every local run mode. + +Its URLs are application-relative proxy paths: its `Configuration` +asks the application to forward the blob path namespace to it, so the +application reverse-proxies byte `PUT`/`GET` here rather than exposing +this server on its own port (see `_proxy.py`). +""" + +import argparse +import asyncio +import grpc +import os +import uvicorn # type: ignore[import] +from rbt.std.blobs.v1.data_plane_pb2 import ( + HTTP_METHOD_GET, + HTTP_METHOD_PUT, + ConfigurationResponse, + DataPlaneBeginUploadResponse, + DataPlaneCompleteUploadResponse, + DataPlaneDeleteResponse, + DataPlaneGetDownloadUrlResponse, + DataPlanePartUploadInstruction, + DataPlaneUploadInstructionsResponse, + ForwardedPath, +) +from rbt.std.blobs.v1.data_plane_pb2_grpc import ( + BlobDataPlaneServicer, + add_BlobDataPlaneServicer_to_server, +) +from reboot.std.blobs.v1._http import build_http_app +from reboot.std.blobs.v1._store import ( + DEFAULT_PART_SIZE_BYTES, + HTTP_PATH_PREFIX, + BlobStoreError, + FilesystemBlobStore, + UploadedPart, +) +from typing import Optional +from uuid import uuid4 + +# The filesystem server binds loopback only: it has no authentication +# (its URLs are HMAC-signed, but the gRPC control surface is not), so +# it must never be reachable off the host. The application reaches it +# over localhost and proxies client byte traffic to it. +LOOPBACK_HOST = "127.0.0.1" + + +class FilesystemDataPlaneServicer(BlobDataPlaneServicer): + """Implements `BlobDataPlane` over a `FilesystemBlobStore`.""" + + def __init__( + self, + store: FilesystemBlobStore, + http_port: int, + ): + self._store = store + self._http_port = http_port + + async def Configuration(self, request, context): + # This server's URLs are application-relative under + # `HTTP_PATH_PREFIX`; ask the application to forward that + # namespace to the HTTP byte endpoint. + return ConfigurationResponse( + part_size=self._store.part_size, + forwarded_paths=[ + ForwardedPath( + method=HTTP_METHOD_GET, + path_prefix=HTTP_PATH_PREFIX + "/", + ), + ForwardedPath( + method=HTTP_METHOD_PUT, + path_prefix=HTTP_PATH_PREFIX + "/", + ), + ], + http_port=self._http_port, + ) + + async def BeginUpload(self, request, context): + upload_id = await self._store.begin_upload( + request.blob_id, + request.content_type, + ) + return DataPlaneBeginUploadResponse(upload_id=upload_id) + + async def UploadInstructions(self, request, context): + instructions = [ + DataPlanePartUploadInstruction( + part_number=part_number, + url=self._store.part_put_url( + request.blob_id, + request.upload_id, + part_number, + ), + ) for part_number in request.part_numbers + ] + return DataPlaneUploadInstructionsResponse(instructions=instructions) + + async def CompleteUpload(self, request, context): + try: + etag = await self._store.complete( + request.blob_id, + request.upload_id, + request.content_type, + [ + UploadedPart( + number=part.number, etag=part.etag, size=part.size + ) for part in request.parts + ], + max_size=( + request.max_size if request.HasField("max_size") else None + ), + ) + return DataPlaneCompleteUploadResponse(etag=etag) + except BlobStoreError as error: + # A permanent failure: report it so the control plane can + # surface it and let the client re-upload. Transient + # failures raise other exceptions, which become gRPC errors + # so the control-plane workflow retries. + return DataPlaneCompleteUploadResponse(error=str(error)) + + async def GetDownloadUrl(self, request, context): + url = self._store.download_url( + request.blob_id, + request.ttl_seconds if request.HasField("ttl_seconds") else None, + ) + return DataPlaneGetDownloadUrlResponse(url=url) + + async def Delete(self, request, context): + await self._store.delete(request.blob_id) + return DataPlaneDeleteResponse() + + +class FilesystemDataPlane: + """A running filesystem blob data plane: the `BlobDataPlane` gRPC + service plus the HTTP byte endpoint its minted URLs point at, both + bound to loopback. Construct via `start()`.""" + + def __init__( + self, + *, + grpc_server: grpc.aio.Server, + grpc_port: int, + http_server, + http_task: asyncio.Task, + http_port: int, + ): + self._grpc_server = grpc_server + self._grpc_port = grpc_port + self._http_server = http_server + self._http_task = http_task + self._http_port = http_port + + @classmethod + async def start( + cls, + *, + directory: str, + part_size: int = DEFAULT_PART_SIZE_BYTES, + grpc_port: int = 0, + http_port: int = 0, + ) -> "FilesystemDataPlane": + """Starts serving, with bytes stored under `directory`. Ports + default to 0 (an ephemeral port chosen by the OS). The HTTP + byte endpoint is brought up before the gRPC surface, so that by + the time `Configuration` is reachable the `http_port` it + reports is already serving.""" + store = FilesystemBlobStore(directory, part_size=part_size) + + http_server = uvicorn.Server( + uvicorn.Config( + build_http_app(store), + host=LOOPBACK_HOST, + port=http_port, + log_level="warning", + ) + ) + http_task = asyncio.create_task(http_server.serve()) + while not http_server.started: + if http_task.done(): + # Startup failed (e.g. the requested port is in use); + # surface the underlying error. + http_task.result() + raise RuntimeError( + "The blob data plane's HTTP server exited during " + "startup" + ) + await asyncio.sleep(0.01) + actual_http_port = http_server.servers[0].sockets[0].getsockname()[1] + + # From here on the HTTP server is live; shut it down if the + # gRPC surface fails to come up, so a failed `start` leaves + # nothing running. + try: + grpc_server = grpc.aio.server() + add_BlobDataPlaneServicer_to_server( + FilesystemDataPlaneServicer(store, actual_http_port), + grpc_server, + ) + actual_grpc_port = grpc_server.add_insecure_port( + f"{LOOPBACK_HOST}:{grpc_port}" + ) + if actual_grpc_port == 0: + raise RuntimeError( + "The blob data plane's gRPC server could not bind " + f"port {grpc_port}" + ) + await grpc_server.start() + except BaseException: + http_server.should_exit = True + await http_task + raise + + return cls( + grpc_server=grpc_server, + grpc_port=actual_grpc_port, + http_server=http_server, + http_task=http_task, + http_port=actual_http_port, + ) + + @property + def grpc_port(self) -> int: + return self._grpc_port + + @property + def http_port(self) -> int: + return self._http_port + + @property + def url(self) -> str: + """The gRPC address to put in `REBOOT_BLOB_DATA_PLANE_URL`.""" + return f"{LOOPBACK_HOST}:{self._grpc_port}" + + async def stop(self) -> None: + await self._grpc_server.stop(grace=None) + self._http_server.should_exit = True + await self._http_task + + async def wait(self) -> None: + """Blocks until the servers terminate (they don't, absent + `stop()`; this is how the standalone program serves forever).""" + await asyncio.gather( + self._grpc_server.wait_for_termination(), + self._http_task, + ) + + +def _write_ready_file(path: str, grpc_port: int, http_port: int) -> None: + """Atomically writes the two chosen ports so the spawning process + learns them only once both servers are listening.""" + temp_path = f"{path}.{uuid4().hex}.tmp" + with open(temp_path, "w") as f: + f.write(f"{grpc_port}\n{http_port}\n") + f.flush() + os.fsync(f.fileno()) + os.replace(temp_path, path) + + +async def serve( + directory: str, + grpc_port: int = 0, + http_port: int = 0, + part_size: int = DEFAULT_PART_SIZE_BYTES, + ready_file: Optional[str] = None, +) -> None: + """Runs the data plane until terminated. When `ready_file` is + given, the actually-bound ports are written to it once both + endpoints are listening, for the spawning process to read.""" + data_plane = await FilesystemDataPlane.start( + directory=directory, + part_size=part_size, + grpc_port=grpc_port, + http_port=http_port, + ) + if ready_file is not None: + _write_ready_file( + ready_file, + data_plane.grpc_port, + data_plane.http_port, + ) + await data_plane.wait() + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Filesystem blob data-plane server." + ) + parser.add_argument("--directory", required=True) + parser.add_argument("--grpc-port", type=int, default=0) + parser.add_argument("--http-port", type=int, default=0) + parser.add_argument( + "--part-size", + type=int, + default=DEFAULT_PART_SIZE_BYTES, + ) + parser.add_argument("--ready-file", default=None) + args = parser.parse_args() + asyncio.run( + serve( + args.directory, + args.grpc_port, + args.http_port, + part_size=args.part_size, + ready_file=args.ready_file, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/reboot/std/blobs/v1/_http.py b/reboot/std/blobs/v1/_http.py new file mode 100644 index 00000000..132e66e2 --- /dev/null +++ b/reboot/std/blobs/v1/_http.py @@ -0,0 +1,183 @@ +"""The HTTP byte endpoint of the filesystem blob data-plane server. + +Serves `PUT` (part upload) and `GET` (download) under +`/__/reboot/blob/`. The filesystem server (`_filesystem_server.py`) +runs this on localhost; the application's `Blob` library reverse- +proxies to it (see `_proxy.py`), so the bytes never leave a single +origin even though they live in a separate process. + +These handlers are self-authorizing: every URL carries an expiring +HMAC signature minted by the data plane, so the handlers never call +back into Reboot state. They touch only the store's directory, +mirroring how a presigned S3 URL is served by S3 without consulting +the application. +""" + +import hashlib +import hmac +import os +import re +import time +from reboot.std.blobs.v1._store import ( + HTTP_PATH_PREFIX, + MAX_PARTS, + FilesystemBlobStore, +) +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import Response, StreamingResponse +from starlette.routing import Route + +_STREAM_CHUNK_SIZE = 1024 * 1024 + +# Path parameters are also filesystem path components; restrict them +# to the alphabets the store actually produces (URL-safe base64 blob +# ids, hex upload ids) as defense in depth against traversal — even +# though a forged path could never carry a valid signature. +_ENCODED_BLOB_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+={0,2}$") +_UPLOAD_ID_PATTERN = re.compile(r"^[0-9a-f]{32}$") + + +class _PartTooLarge(Exception): + pass + + +def _signature_matches(expected: str, actual: str) -> bool: + return hmac.compare_digest(expected, actual) + + +def _expired(request: Request) -> bool: + expiration = request.query_params.get("exp", "0") + return not expiration.isdigit() or int(expiration) < time.time() + + +def _make_put_part(store: FilesystemBlobStore): + + async def put_part(request: Request) -> Response: + blob = request.path_params["blob"] + upload = request.path_params["upload"] + try: + part = int(request.path_params["part"]) + except ValueError: + return Response(status_code=400, content="Invalid part number") + + if part < 1 or part > MAX_PARTS: + return Response(status_code=400, content="Invalid part number") + if ( + not _ENCODED_BLOB_ID_PATTERN.match(blob) or + not _UPLOAD_ID_PATTERN.match(upload) + ): + return Response(status_code=400, content="Invalid blob id") + if _expired(request): + return Response(status_code=403, content="URL expired") + expiration = int(request.query_params.get("exp", "0")) + expected = store.signature_for_put(blob, upload, part, expiration) + if not _signature_matches( + expected, request.query_params.get("sig", "") + ): + return Response(status_code=403, content="Invalid signature") + + path = store.part_path(blob, upload, part) + # The upload directory is created by `begin_upload`; a missing + # directory means the blob was never created (or was deleted). + if not os.path.isdir(os.path.dirname(path)): + return Response(status_code=404, content="No such upload") + + # Refuse to mutate a committed blob's bytes: the part files + # *are* the committed object's on-disk representation, so a + # part-PUT URL minted just before commit must not still be + # usable to tamper with the bytes afterwards. + meta = store.read_meta(blob) + if meta is not None and meta.get("committed", False): + return Response(status_code=409, content="Blob already committed") + + digest = hashlib.md5() + size = 0 + try: + with open(path, "wb") as f: + async for chunk in request.stream(): + if size + len(chunk) > store.part_size: + raise _PartTooLarge() + digest.update(chunk) + size += len(chunk) + f.write(chunk) + f.flush() + os.fsync(f.fileno()) + except _PartTooLarge: + os.unlink(path) + return Response( + status_code=413, + content=( + "Part exceeds the maximum part size of " + f"{store.part_size} bytes" + ), + ) + + # Match S3: the ETag response header is the part's MD5, quoted. + return Response( + status_code=200, + headers={"ETag": f'"{digest.hexdigest()}"'}, + ) + + return put_part + + +def _make_get_blob(store: FilesystemBlobStore): + + async def get_blob(request: Request) -> Response: + blob = request.path_params["blob"] + if not _ENCODED_BLOB_ID_PATTERN.match(blob): + return Response(status_code=400, content="Invalid blob id") + if _expired(request): + return Response(status_code=403, content="URL expired") + expiration = int(request.query_params.get("exp", "0")) + expected = store.signature_for_get(blob, expiration) + if not _signature_matches( + expected, request.query_params.get("sig", "") + ): + return Response(status_code=403, content="Invalid signature") + + meta = store.read_meta(blob) + if meta is None or not meta.get("committed", False): + return Response(status_code=404, content="No such blob") + + upload_id = meta["upload_id"] + parts = meta["parts"] + total_size = sum(part["size"] for part in parts) + + async def stream(): + for part in sorted(parts, key=lambda part: part["number"]): + path = store.part_path(blob, upload_id, part["number"]) + with open(path, "rb") as f: + while chunk := f.read(_STREAM_CHUNK_SIZE): + yield chunk + + return StreamingResponse( + stream(), + media_type=meta["content_type"], + headers={ + "Content-Length": str(total_size), + "ETag": f'"{meta["etag"]}"', + "Accept-Ranges": "none", + }, + ) + + return get_blob + + +def build_http_app(store: FilesystemBlobStore) -> Starlette: + """Builds the Starlette app serving `store`'s byte `PUT`/`GET`.""" + return Starlette( + routes=[ + Route( + HTTP_PATH_PREFIX + "/{blob}/{upload}/parts/{part}", + _make_put_part(store), + methods=["PUT"], + ), + Route( + HTTP_PATH_PREFIX + "/{blob}", + _make_get_blob(store), + methods=["GET"], + ), + ], + ) diff --git a/reboot/std/blobs/v1/_proxy.py b/reboot/std/blobs/v1/_proxy.py new file mode 100644 index 00000000..588c3e82 --- /dev/null +++ b/reboot/std/blobs/v1/_proxy.py @@ -0,0 +1,146 @@ +"""Application-side reverse proxy for a blob data plane's forwarded +paths. + +A data plane whose URLs are not directly reachable by clients (e.g. +the filesystem server, which binds localhost only) asks, via its +`Configuration`, for path namespaces to be forwarded to it. The `Blob` +library then registers these routes on the application's own HTTP +server, forwarding the requests to the data plane's HTTP endpoint. +That keeps the data plane off any externally-exposed port: a single +application origin/tunnel serves both the control plane and the +bytes. + +The proxy is deliberately dumb: it forwards the request path and query +verbatim and never inspects them. The data plane minted the URL and +validates its own signature, so the proxy adds no trust. The one thing +it enforces is the namespace: every forwarded path must live under +`FORWARDED_PATH_PREFIX`, so a data plane can never claim application +routes. +""" + +import aiohttp +from rbt.std.blobs.v1.data_plane_pb2 import ( + HTTP_METHOD_GET, + HTTP_METHOD_PUT, + ForwardedPath, + HttpMethod, +) +from reboot.aio.http import PythonWebFramework +from reboot.std.blobs.v1._data_plane import FORWARDED_PATH_PREFIX +from starlette.requests import Request +from starlette.responses import Response, StreamingResponse +from typing import Iterable + +_STREAM_CHUNK_SIZE = 1024 * 1024 + +# Response headers worth carrying back from the data plane on a +# download; others are hop-by-hop or recomputed by the framework. +_FORWARDED_GET_HEADERS = ( + "Content-Type", + "Content-Length", + "ETag", + "Accept-Ranges", +) + + +def mount_proxy_routes( + http: PythonWebFramework.HTTP, + proxy_target_url: str, + forwarded_paths: Iterable[ForwardedPath], +) -> None: + """Registers a reverse-proxy route for each of the data plane's + `forwarded_paths`, forwarding to `proxy_target_url`. Refuses paths + outside `FORWARDED_PATH_PREFIX` and unknown methods.""" + + target = proxy_target_url.rstrip("/") + + def _forward_url(request: Request) -> str: + url = target + request.url.path + if request.url.query: + url += "?" + request.url.query + return url + + async def put_forward(rest: str, request: Request) -> Response: + session = aiohttp.ClientSession() + try: + upstream = await session.put( + _forward_url(request), + data=request.stream(), + ) + body = await upstream.read() + headers = {} + if "ETag" in upstream.headers: + headers["ETag"] = upstream.headers["ETag"] + return Response( + content=body, + status_code=upstream.status, + headers=headers, + ) + finally: + await session.close() + + async def get_forward(rest: str, request: Request) -> Response: + session = aiohttp.ClientSession() + # Close the session on every non-streaming path (connection + # error, non-200); on the streaming path the generator's + # `finally` closes it once the body is fully read or the client + # disconnects. + try: + upstream = await session.get(_forward_url(request)) + except Exception: + await session.close() + raise + + if upstream.status != 200: + try: + body = await upstream.read() + finally: + await session.close() + return Response(content=body, status_code=upstream.status) + + headers = { + name: upstream.headers[name] + for name in _FORWARDED_GET_HEADERS + if name in upstream.headers + } + + async def stream(): + try: + async for chunk in upstream.content.iter_chunked( + _STREAM_CHUNK_SIZE + ): + yield chunk + finally: + await session.close() + + return StreamingResponse( + stream(), + status_code=200, + media_type=upstream.headers.get("Content-Type"), + headers=headers, + ) + + mounted: set[tuple[int, str]] = set() + for forwarded in forwarded_paths: + prefix = forwarded.path_prefix + if not prefix.startswith(FORWARDED_PATH_PREFIX): + raise ValueError( + f"Blob data plane requested forwarding of '{prefix}', " + f"which is outside `{FORWARDED_PATH_PREFIX}`; refusing" + ) + if (forwarded.method, prefix) in mounted: + continue + mounted.add((forwarded.method, prefix)) + # `{rest:path}` matches anything, including `/`s and the empty + # string, so the route covers exactly "path starts with + # `prefix`". + route = prefix + "{rest:path}" + if forwarded.method == HTTP_METHOD_GET: + http.get(route)(get_forward) + elif forwarded.method == HTTP_METHOD_PUT: + http.put(route)(put_forward) + else: + raise ValueError( + "Blob data plane requested forwarding with unsupported " + f"method {HttpMethod.Name(forwarded.method)}" + ) diff --git a/reboot/std/blobs/v1/_store.py b/reboot/std/blobs/v1/_store.py new file mode 100644 index 00000000..efa58f6b --- /dev/null +++ b/reboot/std/blobs/v1/_store.py @@ -0,0 +1,333 @@ +"""The filesystem blob store: bytes storage for the open-source blob +data plane. + +A blob's *bytes* live here; all its metadata lives in the `Blob` state +machine (the control plane), which talks to the data plane only over +the `BlobDataPlane` gRPC interface (see `data_plane.proto` — that +interface, not this module, is the contract a data plane implements). +The store mimics S3's multipart-upload semantics (numbered parts, +per-part MD5 ETags, ETag-validating completion) so that clients drive +one protocol regardless of which data plane serves them. +""" + +import asyncio +import base64 +import hashlib +import hmac +import json +import os +import shutil +import time +from dataclasses import dataclass +from reboot.crypto import root_keys +from typing import Optional +from uuid import uuid4 + +# The part size clients should use. Every part except the last must be +# exactly this size. Must be at least 5 MiB (the S3 minimum part size, +# mirrored here so that filesystem- and S3-backed data planes are +# interchangeable). +DEFAULT_PART_SIZE_BYTES = 8 * 1024 * 1024 + +# The maximum number of parts in one blob, following S3. +MAX_PARTS = 10000 + +# Default (and maximum) validity of minted upload/download URLs. +DEFAULT_URL_TTL_SECONDS = 15 * 60 +MAX_URL_TTL_SECONDS = 7 * 24 * 60 * 60 + +# The URL path prefix under which blob bytes are `PUT` and `GET`: the +# filesystem data-plane server serves it, and the application's proxy +# routes (see `_proxy.py`) forward it. +HTTP_PATH_PREFIX = "/__/reboot/blob" + +# HKDF `info` (domain separator) for the filesystem store's URL-signing +# key. +_SIGNING_INFO = b"reboot.std.blobs.url-signing" + + +class BlobStoreError(Exception): + """A permanent storage failure (e.g. a part ETag mismatch at + completion time), reported to the control plane as a + `CompleteUpload` `error` so the client can re-upload. Transient + failures (e.g. network errors) are raised as their original + exception types instead, becoming gRPC errors that the control + plane's workflow retries.""" + + +@dataclass(frozen=True) +class UploadedPart: + """One part of an upload, as reported by the client.""" + number: int + etag: str + size: int + + +def _encode_blob_id(blob_id: str) -> str: + """Encodes a blob id into a string safe for use as both a directory + name and a URL path segment.""" + return base64.urlsafe_b64encode(blob_id.encode()).decode() + + +def _fsync_path(path: str) -> None: + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +class FilesystemBlobStore: + """Stores blob bytes as part files on the local filesystem, served + over HTTP by the filesystem data-plane server (see `_http.py`). + + Layout, under `directory`: + + {encoded_blob_id}/ + meta.json Content type; part manifest and + composite ETag once committed. + {upload_id}/ + part.{number:08d} One file per uploaded part. + + Parts are written once under a random `upload_id` directory (so no + temp-file-and-rename protocol is needed) and fsynced before the + data plane returns their ETag. The part files remain the committed + object's on-disk representation: downloads stream them in part + order, so completion never rewrites bytes. + """ + + def __init__( + self, + directory: str, + part_size: int = DEFAULT_PART_SIZE_BYTES, + ): + self._directory = directory + self._part_size = part_size + os.makedirs(directory, exist_ok=True) + + @property + def directory(self) -> str: + return self._directory + + @property + def part_size(self) -> int: + return self._part_size + + def _signing_key(self) -> bytes: + """The URL-signing key, derived from the active version of the + Reboot-managed cryptographic root keys (see + `reboot.crypto.root_keys`) with a blob-specific domain + separator. Rotating the root keys therefore invalidates + outstanding URLs; that is acceptable because URLs are + short-lived and clients can always mint fresh ones.""" + return root_keys.derive_key( + info=_SIGNING_INFO, + version=root_keys.active_version(), + ) + + def _sign(self, *parts: str) -> str: + message = "\n".join(parts).encode() + return hmac.new(self._signing_key(), message, + hashlib.sha256).hexdigest() + + def signature_for_put( + self, + encoded_blob_id: str, + upload_id: str, + part_number: int, + expiration: int, + ) -> str: + return self._sign( + "PUT", encoded_blob_id, upload_id, str(part_number), + str(expiration) + ) + + def signature_for_get( + self, + encoded_blob_id: str, + expiration: int, + ) -> str: + return self._sign("GET", encoded_blob_id, str(expiration)) + + def blob_directory(self, encoded_blob_id: str) -> str: + return os.path.join(self._directory, encoded_blob_id) + + def part_path( + self, + encoded_blob_id: str, + upload_id: str, + part_number: int, + ) -> str: + return os.path.join( + self.blob_directory(encoded_blob_id), + upload_id, + f"part.{part_number:08d}", + ) + + def _meta_path(self, encoded_blob_id: str) -> str: + return os.path.join(self.blob_directory(encoded_blob_id), "meta.json") + + def read_meta(self, encoded_blob_id: str) -> Optional[dict]: + try: + with open(self._meta_path(encoded_blob_id), "r") as f: + return json.load(f) + except FileNotFoundError: + return None + + def _write_meta(self, encoded_blob_id: str, meta: dict) -> None: + # Write to a temp file and atomically rename, so a crash + # mid-write can never leave a torn `meta.json` that a + # concurrent `read_meta` would fail to parse. + path = self._meta_path(encoded_blob_id) + temp_path = f"{path}.{uuid4().hex}.tmp" + with open(temp_path, "w") as f: + json.dump(meta, f) + f.flush() + os.fsync(f.fileno()) + os.replace(temp_path, path) + _fsync_path(os.path.dirname(path)) + + async def begin_upload(self, blob_id: str, content_type: str) -> str: + encoded = _encode_blob_id(blob_id) + + def sync(): + # Idempotent by blob id: if an uncommitted session already + # exists (a retried `BeginUpload`), reuse it rather than + # orphaning it under a fresh upload id. + existing = self.read_meta(encoded) + if ( + existing is not None and + not existing.get("committed", False) and + "upload_id" in existing + ): + upload_id = existing["upload_id"] + reuse = True + else: + upload_id = uuid4().hex + reuse = False + # Create the upload directory (and, with it, the blob + # directory) before writing `meta.json` into the latter. + os.makedirs( + os.path.join(self.blob_directory(encoded), upload_id), + exist_ok=True, + ) + if not reuse: + self._write_meta( + encoded, + { + "content_type": content_type, + "committed": False, + "upload_id": upload_id, + }, + ) + return upload_id + + return await asyncio.to_thread(sync) + + def part_put_url( + self, + blob_id: str, + upload_id: str, + part_number: int, + ) -> str: + encoded = _encode_blob_id(blob_id) + expiration = int(time.time()) + DEFAULT_URL_TTL_SECONDS + signature = self.signature_for_put( + encoded, upload_id, part_number, expiration + ) + return ( + f"{HTTP_PATH_PREFIX}/{encoded}/{upload_id}/parts/{part_number}" + f"?exp={expiration}&sig={signature}" + ) + + async def complete( + self, + blob_id: str, + upload_id: str, + content_type: str, + parts: list[UploadedPart], + max_size: Optional[int] = None, + ) -> str: + + def sync(): + encoded = _encode_blob_id(blob_id) + digests = [] + manifest = [] + total_size = 0 + for part in sorted(parts, key=lambda part: part.number): + path = self.part_path(encoded, upload_id, part.number) + digest = hashlib.md5() + size = 0 + try: + with open(path, "rb") as f: + while chunk := f.read(1024 * 1024): + digest.update(chunk) + size += len(chunk) + except FileNotFoundError: + raise BlobStoreError( + f"part {part.number} was never uploaded" + ) + if digest.hexdigest() != part.etag.strip('"'): + raise BlobStoreError( + f"part {part.number} ETag mismatch: the uploaded " + "bytes do not match what was reported via " + "`PartUploaded`" + ) + if size != part.size: + raise BlobStoreError( + f"part {part.number} size mismatch: uploaded " + f"{size} bytes but {part.size} were reported via " + "`PartUploaded`" + ) + total_size += size + digests.append(digest.digest()) + manifest.append({"number": part.number, "size": size}) + + # Verify the *real* total against `max_size` (not the + # already-checked reported sizes) as defense in depth. + if max_size is not None and total_size > max_size: + raise BlobStoreError( + f"uploaded {total_size} bytes exceeds the maximum of " + f"{max_size}" + ) + + # Composite ETag, S3-style: the MD5 of the concatenated + # part MD5 digests, suffixed with the part count. + etag = ( + hashlib.md5(b"".join(digests)).hexdigest() + f"-{len(digests)}" + ) + self._write_meta( + encoded, + { + "content_type": content_type, + "committed": True, + "etag": etag, + "upload_id": upload_id, + "parts": manifest, + }, + ) + return etag + + return await asyncio.to_thread(sync) + + def download_url( + self, + blob_id: str, + ttl_seconds: Optional[int] = None, + ) -> str: + encoded = _encode_blob_id(blob_id) + ttl = min( + ttl_seconds or DEFAULT_URL_TTL_SECONDS, + MAX_URL_TTL_SECONDS, + ) + expiration = int(time.time()) + ttl + signature = self.signature_for_get(encoded, expiration) + return f"{HTTP_PATH_PREFIX}/{encoded}?exp={expiration}&sig={signature}" + + async def delete(self, blob_id: str) -> None: + encoded = _encode_blob_id(blob_id) + + def sync(): + shutil.rmtree(self.blob_directory(encoded), ignore_errors=True) + + await asyncio.to_thread(sync) diff --git a/reboot/std/blobs/v1/blobs.py b/reboot/std/blobs/v1/blobs.py new file mode 100644 index 00000000..c773bb71 --- /dev/null +++ b/reboot/std/blobs/v1/blobs.py @@ -0,0 +1,611 @@ +"""Blob storage: large binary objects whose bytes live outside state. + +A `Blob` is the *control plane* for one immutable-once-committed +binary object: its state holds only metadata (content type, size, +upload progress, lifecycle status) while the bytes live in a *data +plane* — a `BlobDataPlane` gRPC service (see `data_plane.proto`), +discovered via `REBOOT_BLOB_DATA_PLANE_URL` — and travel directly +between the client and that data plane via URLs minted by +`UploadInstructions` and `DownloadUrl`. Uploads are resumable: parts +are idempotent by number, so a client that lost its connection +re-fetches instructions and re-uploads whatever `Info` does not yet +report. + +Authorization model: blob *creation* is application-mediated — only +application code may call `Create`, which is where size and quota +policy belongs (enforced directly or via `max_size`). The blob's +framework-generated random id then acts as a capability. Upload-side +calls (`UploadInstructions`, `PartUploaded`, `Commit`) and `Remove` are +restricted to the `uploader_id` recorded at `Create` — unless +`uploader_id` is left empty, which deliberately allows anyone who knows +the blob's id to upload (for applications without end-user +authentication). Downloads (`DownloadUrl`) are open to anyone who knows +the id by default, but if `Create` (or a later `SetDownloaders`) +records a `downloader_ids` allow-list only the listed users may download; +an empty list restricts downloads to app-internal callers, and the +uploader is *not* implicitly a downloader. `Info` (metadata and upload +progress, watchable reactively) is visible to anyone who may upload or +download the blob: the `uploader_id` and listed `downloader_ids`, plus +anyone who knows the id whenever either side is left open. +""" + +import asyncio +import log.log +import rbt.v1alpha1.errors_pb2 +import re +from datetime import timedelta +from grpc.aio import AioRpcError +from rbt.std.blobs.v1.blobs_rbt import ( + AlreadyCommitted, + BeginUploadRequest, + BeginUploadResponse, + Blob, + BlobPart, + CommitRequest, + CommitResponse, + CompleteUploadRequest, + CompleteUploadResponse, + CreateRequest, + CreateResponse, + DownloadUrlRequest, + DownloadUrlResponse, + ExpireIfNotCommittedRequest, + ExpireIfNotCommittedResponse, + IncompleteParts, + InfoRequest, + InfoResponse, + NotCommitted, + PartUploadedRequest, + PartUploadedResponse, + PartUploadInstruction, + PerformRemoveRequest, + PerformRemoveResponse, + RemoveRequest, + RemoveResponse, + SetDownloadersRequest, + SetDownloadersResponse, + SizeMismatch, + UploadInstructionsRequest, + UploadInstructionsResponse, +) +from rbt.std.blobs.v1.data_plane_pb2 import ( + ConfigurationRequest, + ConfigurationResponse, + DataPlaneBeginUploadRequest, + DataPlaneCompleteUploadRequest, + DataPlaneDeleteRequest, + DataPlaneGetDownloadUrlRequest, + DataPlaneUploadedPart, + DataPlaneUploadInstructionsRequest, +) +from rbt.std.blobs.v1.data_plane_pb2_grpc import BlobDataPlaneStub +from reboot.aio.applications import Application, Library +from reboot.aio.auth.authorizers import allow_if, is_app_internal +from reboot.aio.contexts import ReaderContext, WorkflowContext, WriterContext +from reboot.aio.http import PythonWebFramework +from reboot.aio.workflows import at_least_once_per_workflow +from reboot.std.blobs.v1._data_plane import ( + ENVVAR_BLOB_DATA_PLANE_URL, + proxy_target_from_environment, + stub_from_environment, +) +from reboot.std.blobs.v1._proxy import mount_proxy_routes +from reboot.std.blobs.v1._store import MAX_PARTS +from typing import Optional + +logger = log.log.get_logger(__name__) + +# How long to keep retrying `Configuration` while the data plane +# comes up, before `pre_run` gives up. The data plane is normally +# already running (spawned by `rbt` or a ready facilitator), so this is +# only a startup-race cushion. +_CONFIGURATION_RETRY_SECONDS = 30 +_CONFIGURATION_RETRY_INTERVAL_SECONDS = 0.5 + +# How long an upload may remain uncommitted before the blob is +# expunged by the `ExpireIfNotCommitted` task `Create` schedules. +DEFAULT_UPLOAD_EXPIRATION = timedelta(hours=24) + +# A part ETag is a 32-character MD5 hex digest (optionally quoted by +# the client, which we strip). The data-plane contract requires a +# valid ETag for each reported part, so a malformed one is rejected +# up front rather than being handed to a data plane. +_PART_ETAG_PATTERN = re.compile(r'^"?[0-9a-fA-F]{32}"?$') + + +def _uploader_or_open(*, context, state=None, request=None, **kwargs): + """Allow app-internal callers and the blob's recorded uploader to + make upload-side calls, or anyone when no uploader was recorded. An + empty `uploader_id` means the blob was created without end-user + authentication, so anyone who knows the blob's id may upload into + it. This handles the app-internal case itself (rather than + composing `is_app_internal` via `any=[...]`) so that an + unauthenticated non-uploader still surfaces as `Unauthenticated` + rather than `PermissionDenied`.""" + if context.app_internal: + return rbt.v1alpha1.errors_pb2.Ok() + if state is None: + return rbt.v1alpha1.errors_pb2.PermissionDenied() + if state.uploader_id == "": + return rbt.v1alpha1.errors_pb2.Ok() + if context.auth is None or context.auth.user_id is None: + return rbt.v1alpha1.errors_pb2.Unauthenticated() + if context.auth.user_id == state.uploader_id: + return rbt.v1alpha1.errors_pb2.Ok() + return rbt.v1alpha1.errors_pb2.PermissionDenied() + + +def _downloader_or_open(*, context, state=None, request=None, **kwargs): + """Allow app-internal callers, and gate `DownloadUrl` on the blob's + download allow-list. When no `downloader_ids` list was recorded (the + field is unset) anyone who knows the blob's id may download; when + one was recorded only the listed users may (an empty list means no + one but app-internal callers). Like `_uploader_or_open`, this + handles app-internal itself so that an unauthenticated non-listed + caller surfaces as `Unauthenticated` rather than + `PermissionDenied`.""" + if context.app_internal: + return rbt.v1alpha1.errors_pb2.Ok() + if state is None: + return rbt.v1alpha1.errors_pb2.PermissionDenied() + if not state.HasField("downloader_ids"): + return rbt.v1alpha1.errors_pb2.Ok() + if context.auth is None or context.auth.user_id is None: + return rbt.v1alpha1.errors_pb2.Unauthenticated() + if context.auth.user_id in state.downloader_ids.user_ids: + return rbt.v1alpha1.errors_pb2.Ok() + return rbt.v1alpha1.errors_pb2.PermissionDenied() + + +class BlobServicer(Blob.Servicer): + + # The data-plane gRPC stub and the part size it reported, both set + # by `BlobsLibrary` once it has connected to the data plane. + _data_plane: BlobDataPlaneStub + _part_size: int + + def authorizer(self): + # Every method is listed explicitly so none can be + # accidentally left ungated. `Create`, `SetDownloaders`, and + # the internal workflow methods are app-internal only; the + # upload-side methods and `Remove` require the caller to be the + # blob's uploader, unless `uploader_id` is empty; `DownloadUrl` + # is gated on the `downloader_ids` allow-list, or open when none + # was recorded; `Info` is visible to anyone who may upload or + # download the blob (either predicate suffices). + return Blob.Authorizer( + create=allow_if(any=[is_app_internal]), + set_downloaders=allow_if(any=[is_app_internal]), + begin_upload=allow_if(any=[is_app_internal]), + complete_upload=allow_if(any=[is_app_internal]), + perform_remove=allow_if(any=[is_app_internal]), + expire_if_not_committed=allow_if(any=[is_app_internal]), + info=allow_if(any=[_uploader_or_open, _downloader_or_open]), + download_url=allow_if(any=[_downloader_or_open]), + upload_instructions=allow_if(any=[_uploader_or_open]), + part_uploaded=allow_if(any=[_uploader_or_open]), + commit=allow_if(any=[_uploader_or_open]), + remove=allow_if(any=[_uploader_or_open]), + ) + + async def create( + self, + context: WriterContext, + request: CreateRequest, + ) -> CreateResponse: + self.state.status = Blob.State.UPLOADING + self.state.content_type = request.content_type + self.state.uploader_id = request.uploader_id + if request.HasField("downloader_ids"): + self.state.downloader_ids.CopyFrom(request.downloader_ids) + if request.HasField("size"): + self.state.size = request.size + if request.HasField("max_size"): + self.state.max_size = request.max_size + + # The storage-backend side effect (provisioning the upload + # session) happens in the `BeginUpload` workflow, not here. + await self.ref().schedule().begin_upload(context) + + # Expunge this blob if it is never committed. + await self.ref().schedule( + when=DEFAULT_UPLOAD_EXPIRATION, + ).expire_if_not_committed(context) + + return CreateResponse() + + async def set_downloaders( + self, + context: WriterContext, + request: SetDownloadersRequest, + ) -> SetDownloadersResponse: + # Replace semantics: a present `downloader_ids` (even empty) + # restricts downloads to the listed users; an omitted one + # removes any restriction so anyone who knows the id may + # download again. + if request.HasField("downloader_ids"): + self.state.downloader_ids.CopyFrom(request.downloader_ids) + else: + self.state.ClearField("downloader_ids") + return SetDownloadersResponse() + + @classmethod + async def begin_upload( + cls, + context: WorkflowContext, + request: BeginUploadRequest, + ) -> BeginUploadResponse: + state = await Blob.ref().read(context) + + async def provision() -> str: + response = await cls._data_plane.BeginUpload( + DataPlaneBeginUploadRequest( + blob_id=context.state_id, + content_type=state.content_type, + ) + ) + return response.upload_id + + upload_id = await at_least_once_per_workflow( + "provision upload session", context, provision + ) + + async def record(state: Blob.State) -> None: + state.upload_id = upload_id + + await Blob.ref().write(context, record) + return BeginUploadResponse() + + async def upload_instructions( + self, + context: ReaderContext, + request: UploadInstructionsRequest, + ) -> UploadInstructionsResponse: + if self.state.status != Blob.State.UPLOADING: + raise Blob.UploadInstructionsAborted(AlreadyCommitted()) + + if not self.state.HasField("upload_id"): + return UploadInstructionsResponse( + ready=False, + part_size=self._part_size, + ) + + part_numbers = [ + number for number in request.part_numbers + if 1 <= number <= MAX_PARTS + ] + response = await self._data_plane.UploadInstructions( + DataPlaneUploadInstructionsRequest( + blob_id=context.state_id, + upload_id=self.state.upload_id, + part_numbers=part_numbers, + ) + ) + instructions = [ + PartUploadInstruction( + part_number=instruction.part_number, + url=instruction.url, + ) for instruction in response.instructions + ] + + return UploadInstructionsResponse( + ready=True, + part_size=self._part_size, + instructions=instructions, + ) + + async def part_uploaded( + self, + context: WriterContext, + request: PartUploadedRequest, + ) -> PartUploadedResponse: + if self.state.status != Blob.State.UPLOADING: + raise Blob.PartUploadedAborted(AlreadyCommitted()) + + # Reject out-of-range part numbers: a bogus record (e.g. the + # proto default `0` from an omitted field) can only be + # overwritten, never removed, so it would make `Commit`'s + # contiguity check fail forever. + if request.part_number < 1 or request.part_number > MAX_PARTS: + raise Blob.PartUploadedAborted(IncompleteParts()) + + # Validate the ETag as an MD5 hex digest, as the data-plane + # contract requires, so a client can't smuggle arbitrary + # content into the value a data plane later relies on to + # finalize the object. + if not _PART_ETAG_PATTERN.match(request.etag): + raise Blob.PartUploadedAborted(IncompleteParts()) + + part = BlobPart( + number=request.part_number, + etag=request.etag, + size=request.size, + ) + + # Safe to call multiple times for the same part number: a + # re-uploaded part overwrites its previous record. + parts = [p for p in self.state.parts if p.number != part.number] + parts.append(part) + parts.sort(key=lambda part: part.number) + + total = sum(part.size for part in parts) + if self.state.HasField("max_size") and total > self.state.max_size: + raise Blob.PartUploadedAborted(SizeMismatch(bytes_uploaded=total)) + + del self.state.parts[:] + self.state.parts.extend(parts) + return PartUploadedResponse() + + async def commit( + self, + context: WriterContext, + request: CommitRequest, + ) -> CommitResponse: + if self.state.status == Blob.State.COMMITTING: + # Idempotent: the `CompleteUpload` workflow is already + # scheduled. + return CommitResponse() + if self.state.status != Blob.State.UPLOADING: + raise Blob.CommitAborted(AlreadyCommitted()) + + numbers = [part.number for part in self.state.parts] + if not numbers or numbers != list(range(1, len(numbers) + 1)): + raise Blob.CommitAborted(IncompleteParts()) + + total = sum(part.size for part in self.state.parts) + if self.state.HasField("size") and total != self.state.size: + raise Blob.CommitAborted(SizeMismatch(bytes_uploaded=total)) + if self.state.HasField("max_size") and total > self.state.max_size: + raise Blob.CommitAborted(SizeMismatch(bytes_uploaded=total)) + + self.state.status = Blob.State.COMMITTING + # Clear any error from a previous failed commit attempt, so a + # client watching `Info` doesn't observe the stale error while + # this fresh attempt is in flight. + self.state.ClearField("commit_error") + await self.ref().schedule().complete_upload(context) + return CommitResponse() + + @classmethod + async def complete_upload( + cls, + context: WorkflowContext, + request: CompleteUploadRequest, + ) -> CompleteUploadResponse: + state = await Blob.ref().read(context) + + # A concurrent `Remove` may have moved the blob out of + # COMMITTING (deletion always wins); if so, don't finalize. + if state.status != Blob.State.COMMITTING: + return CompleteUploadResponse() + + complete_request = DataPlaneCompleteUploadRequest( + blob_id=context.state_id, + upload_id=state.upload_id, + content_type=state.content_type, + parts=[ + DataPlaneUploadedPart( + number=part.number, etag=part.etag, size=part.size + ) for part in state.parts + ], + ) + if state.HasField("max_size"): + complete_request.max_size = state.max_size + + async def attempt() -> tuple: + # A response `error` is a *permanent* failure (e.g. an ETag + # mismatch): report it back onto the blob so the client can + # re-upload and re-commit. A gRPC error is transient and + # propagates, so the workflow retries. + response = await cls._data_plane.CompleteUpload(complete_request) + if response.HasField("error"): + return ("failed", response.error) + return ("committed", response.etag) + + outcome, detail = await at_least_once_per_workflow( + "complete upload", context, attempt + ) + + # Only transition if the blob is still COMMITTING: a + # concurrent `Remove` may have moved it to DELETING/DELETED, + # which must win (otherwise we'd resurrect a deleted blob or + # mark a bytes-less blob COMMITTED). + superseded = [False] + + async def record(state: Blob.State) -> None: + if state.status != Blob.State.COMMITTING: + superseded[0] = True + return + if outcome == "committed": + state.status = Blob.State.COMMITTED + state.etag = detail + state.ClearField("commit_error") + else: + state.status = Blob.State.UPLOADING + state.commit_error = detail + + await Blob.ref().write(context, record) + + # If a delete raced ahead of a successful completion, the + # bytes we just finalized are now orphaned; clean them up. + if superseded[0] and outcome == "committed": + + async def cleanup() -> None: + await cls._data_plane.Delete( + DataPlaneDeleteRequest(blob_id=context.state_id) + ) + + await at_least_once_per_workflow( + "cleanup orphaned bytes", context, cleanup + ) + + return CompleteUploadResponse() + + async def info( + self, + context: ReaderContext, + request: InfoRequest, + ) -> InfoResponse: + response = InfoResponse( + status=self.state.status, + content_type=self.state.content_type, + uploader_id=self.state.uploader_id, + bytes_uploaded=sum(part.size for part in self.state.parts), + parts=self.state.parts, + ) + if self.state.HasField("size"): + response.size = self.state.size + if self.state.HasField("max_size"): + response.max_size = self.state.max_size + if self.state.HasField("etag"): + response.etag = self.state.etag + if self.state.HasField("commit_error"): + response.commit_error = self.state.commit_error + return response + + async def download_url( + self, + context: ReaderContext, + request: DownloadUrlRequest, + ) -> DownloadUrlResponse: + if self.state.status != Blob.State.COMMITTED: + raise Blob.DownloadUrlAborted(NotCommitted()) + download_request = DataPlaneGetDownloadUrlRequest( + blob_id=context.state_id, + ) + if request.HasField("ttl_seconds"): + download_request.ttl_seconds = request.ttl_seconds + response = await self._data_plane.GetDownloadUrl(download_request) + return DownloadUrlResponse(url=response.url) + + async def remove( + self, + context: WriterContext, + request: RemoveRequest, + ) -> RemoveResponse: + if self.state.status in ( + Blob.State.DELETING, + Blob.State.DELETED, + ): + return RemoveResponse() + self.state.status = Blob.State.DELETING + await self.ref().schedule().perform_remove(context) + return RemoveResponse() + + @classmethod + async def perform_remove( + cls, + context: WorkflowContext, + request: PerformRemoveRequest, + ) -> PerformRemoveResponse: + + async def remove() -> None: + await cls._data_plane.Delete( + DataPlaneDeleteRequest(blob_id=context.state_id) + ) + + await at_least_once_per_workflow("remove bytes", context, remove) + + async def record(state: Blob.State) -> None: + state.status = Blob.State.DELETED + del state.parts[:] + + await Blob.ref().write(context, record) + return PerformRemoveResponse() + + async def expire_if_not_committed( + self, + context: WriterContext, + request: ExpireIfNotCommittedRequest, + ) -> ExpireIfNotCommittedResponse: + if self.state.status == Blob.State.UPLOADING: + self.state.status = Blob.State.DELETING + await self.ref().schedule().perform_remove(context) + elif self.state.status == Blob.State.COMMITTING: + # A commit is in flight. If it fails it will revert to + # UPLOADING and could then be abandoned, so re-arm the + # expiration check rather than dropping it. + await self.ref().schedule( + when=DEFAULT_UPLOAD_EXPIRATION, + ).expire_if_not_committed(context) + return ExpireIfNotCommittedResponse() + + +BLOBS_LIBRARY_NAME = "reboot.std.blobs.v1.blobs" + + +class BlobsLibrary(Library): + name = BLOBS_LIBRARY_NAME + + def __init__(self): + self._connected = False + + def servicers(self): + return [BlobServicer] + + async def pre_run(self, application: Application) -> None: + # `pre_run` may be called more than once (e.g. a test that + # `up`s an application after a `down`); connect once. + if self._connected: + return + + stub = stub_from_environment() + BlobServicer._data_plane = stub + + configuration = await self._configuration(stub) + BlobServicer._part_size = configuration.part_size + + # A data plane whose URLs are not directly reachable (e.g. the + # localhost filesystem server) asks for paths to be forwarded + # to it, and the application proxies those to its HTTP + # endpoint; one that serves its own URLs (e.g. S3) asks for + # none and needs no application routes. + if configuration.forwarded_paths: + if not isinstance(application.web_framework, PythonWebFramework): + # Better to fail fast here than to hand out URLs that + # will 404: without the proxy routes, a forwarded-path + # data plane cannot serve any bytes. + raise RuntimeError( + "This blob data plane needs paths forwarded to it, " + "which only Python applications currently support; " + "configure a data plane whose URLs are directly " + "reachable by clients via " + f"`{ENVVAR_BLOB_DATA_PLANE_URL}`." + ) + mount_proxy_routes( + application.http, + proxy_target_from_environment(configuration.http_port), + configuration.forwarded_paths, + ) + + self._connected = True + + async def _configuration( + self, + stub: BlobDataPlaneStub, + ) -> ConfigurationResponse: + # The data plane is normally already running, but tolerate a + # startup race by retrying while it becomes reachable. + attempts = int( + _CONFIGURATION_RETRY_SECONDS / + _CONFIGURATION_RETRY_INTERVAL_SECONDS + ) + last_error: Optional[AioRpcError] = None + for _ in range(attempts): + try: + return await stub.Configuration(ConfigurationRequest()) + except AioRpcError as error: + last_error = error + await asyncio.sleep(_CONFIGURATION_RETRY_INTERVAL_SECONDS) + raise RuntimeError( + "Timed out waiting for the blob data plane to become " + f"reachable via `{ENVVAR_BLOB_DATA_PLANE_URL}`." + ) from last_error + + +def servicers(): + return [BlobServicer] + + +def blobs_library() -> BlobsLibrary: + return BlobsLibrary() diff --git a/reboot/std/blobs/v1/index.ts b/reboot/std/blobs/v1/index.ts new file mode 100644 index 00000000..3a1054bb --- /dev/null +++ b/reboot/std/blobs/v1/index.ts @@ -0,0 +1,30 @@ +import { NativeLibrary, NativeServicer } from "@reboot-dev/reboot"; + +export * from "@reboot-dev/reboot-std-api/blobs/v1/blobs_rbt.js"; + +// The servicers are implemented in Python (the data-plane client +// lives there); Node.js applications host them as "native" servicers. +// +// NOTE: the application-side routes that proxy bytes to a data plane +// requesting forwarded paths (such as the local filesystem one) are +// currently only registered by Python applications; Node.js +// applications need a data plane whose URLs are directly reachable by +// clients (no forwarded paths, e.g. Reboot Cloud's). +export default { + servicers: (): NativeServicer[] => { + return [ + { + nativeServicerModule: "reboot.std.blobs.v1.blobs", + }, + ]; + }, +}; + +export const BLOBS_LIBRARY_NAME = "reboot.std.blobs.v1.blobs"; + +export function blobsLibrary(): NativeLibrary { + return { + nativeLibraryModule: "reboot.std.blobs.v1.blobs", + nativeLibraryFunction: "blobs_library", + }; +} diff --git a/reboot/std/blobs/v1/package.json b/reboot/std/blobs/v1/package.json new file mode 100644 index 00000000..3dbc1ca5 --- /dev/null +++ b/reboot/std/blobs/v1/package.json @@ -0,0 +1,3 @@ +{ + "type": "module" +} diff --git a/reboot/std/react/BUILD.bazel b/reboot/std/react/BUILD.bazel index 9e43a0a8..24da2f34 100644 --- a/reboot/std/react/BUILD.bazel +++ b/reboot/std/react/BUILD.bazel @@ -48,6 +48,7 @@ ts_project( }, visibility = ["//visibility:public"], deps = [ + "//reboot/std/react/blobs:blobs_ts", "//reboot/std/react/presence:presence_ts", ], ) diff --git a/reboot/std/react/blobs/BUILD.bazel b/reboot/std/react/blobs/BUILD.bazel new file mode 100644 index 00000000..6c7803e1 --- /dev/null +++ b/reboot/std/react/blobs/BUILD.bazel @@ -0,0 +1,27 @@ +load("@aspect_rules_ts//ts:defs.bzl", "ts_project") + +ts_project( + name = "blobs_ts", + srcs = [ + "index.tsx", + "package.json", + ], + declaration = True, + tsconfig = { + "compilerOptions": { + "declaration": True, + "jsx": "react-jsx", + "module": "nodenext", + "moduleResolution": "nodenext", + "target": "es2020", + }, + }, + visibility = ["//visibility:public"], + deps = [ + "//:node_modules/@reboot-dev/reboot-api", + "//:node_modules/@reboot-dev/reboot-react", + "//:node_modules/@reboot-dev/reboot-std-api", + "//:node_modules/@reboot-dev/reboot-web", + "//:node_modules/react", + ], +) diff --git a/reboot/std/react/blobs/index.tsx b/reboot/std/react/blobs/index.tsx new file mode 100644 index 00000000..cc30e985 --- /dev/null +++ b/reboot/std/react/blobs/index.tsx @@ -0,0 +1,309 @@ +// Browser-side helpers for `reboot.std.blobs`: a dead-simple hook for +// uploading a `File` into a `Blob` the application backend has +// created, plus the lower-level `BlobUploader` for bytes that come +// from somewhere other than a file input. +// +// The control-plane calls (`UploadInstructions`, `PartUploaded`, +// `Commit`, `Info`, `DownloadUrl`) go over the ordinary Reboot HTTP +// RPC path; the bytes themselves are `PUT` directly to the URLs the +// control plane mints (the application's own data plane for the +// `filesystem` store, presigned S3 URLs for the `s3` store — the +// uploader neither knows nor cares which). + +import { stateIdToRef } from "@reboot-dev/reboot-api"; +import { useRebootClient } from "@reboot-dev/reboot-react"; +import { useEffect, useMemo, useState } from "react"; + +// Re-exported so applications can reactively render blob metadata +// (e.g. a progress bar for an attachment some *other* client is +// uploading) without a separate import of the generated client. +export { useBlob } from "@reboot-dev/reboot-std-api/blobs/v1/blobs_rbt_react.js"; + +const STATE_TYPE = "rbt.std.blobs.v1.Blob"; +const METHODS_SERVICE = "rbt.std.blobs.v1.BlobMethods"; + +export interface UploadProgress { + uploadedBytes: number; + totalBytes: number; +} + +export interface UploadOptions { + onProgress?: (progress: UploadProgress) => void; + signal?: AbortSignal; +} + +export interface UploadResult { + etag?: string; + error?: string; +} + +async function sleep(milliseconds: number): Promise { + await new Promise((resolve) => setTimeout(resolve, milliseconds)); +} + +async function callBlobMethod( + options: { url: string; blobId: string; bearerToken?: string }, + method: string, + request: object +): Promise { + const headers: Record = { + "Content-Type": "application/json", + }; + if (options.bearerToken !== undefined) { + headers["Authorization"] = `Bearer ${options.bearerToken}`; + } + const response = await fetch( + `${options.url}/__/reboot/rpc/${stateIdToRef( + STATE_TYPE, + options.blobId + )}/` + `${METHODS_SERVICE}/${method}`, + { + method: "POST", + headers, + body: JSON.stringify(request), + } + ); + if (!response.ok) { + throw new Error( + `Blob.${method} failed (${response.status}): ` + + `${await response.text()}` + ); + } + return await response.json(); +} + +/** + * Uploads bytes into a `Blob` that the application backend has + * created (blob creation is always application-mediated; ask your + * backend for a blob id first). + * + * Use `upload(...)` for a `File`/`Blob`/`Uint8Array` you already + * have, or `putPart(...)`/`commit()` directly when producing bytes + * incrementally from some other source. Attaching an uploader to a + * partially-uploaded blob resumes it: already-confirmed parts are + * skipped. + */ +export class BlobUploader { + private options: { url: string; blobId: string; bearerToken?: string }; + private confirmed: Map = new Map(); + + constructor(options: { url: string; blobId: string; bearerToken?: string }) { + this.options = options; + } + + private async call(method: string, request: object): Promise { + return await callBlobMethod(this.options, method, request); + } + + /** + * Fetches upload instructions for the given part numbers, waiting + * for the blob's upload session to be provisioned. + */ + async instructions( + partNumbers: number[], + options?: { signal?: AbortSignal } + ): Promise<{ partSize: number; urls: Map }> { + while (true) { + options?.signal?.throwIfAborted(); + const response = await this.call("UploadInstructions", { + partNumbers, + }); + if (response.ready) { + const urls = new Map(); + for (const instruction of response.instructions ?? []) { + urls.set( + instruction.partNumber, + new URL(instruction.url, this.options.url).toString() + ); + } + return { partSize: Number(response.partSize), urls }; + } + await sleep(100); + } + } + + /** + * `PUT`s one part's bytes to the data plane and reports it to the + * control plane. Idempotent per part number. + */ + async putPart( + partNumber: number, + bytes: globalThis.Blob | Uint8Array, + options?: { signal?: AbortSignal } + ): Promise { + const { urls } = await this.instructions([partNumber], options); + const url = urls.get(partNumber); + if (url === undefined) { + throw new Error(`No upload URL for part ${partNumber}`); + } + const response = await fetch(url, { + method: "PUT", + body: bytes, + signal: options?.signal, + }); + if (!response.ok) { + throw new Error( + `Part ${partNumber} upload failed (${response.status}): ` + + `${await response.text()}` + ); + } + const etag = (response.headers.get("ETag") ?? "").replace(/"/g, ""); + if (etag === "") { + throw new Error( + `Part ${partNumber} upload returned no ETag; if this ` + + "application uses an S3-compatible store, its bucket CORS " + + "configuration must expose the `ETag` header" + ); + } + const size = bytes instanceof Uint8Array ? bytes.byteLength : bytes.size; + await this.call("PartUploaded", { + partNumber, + etag, + size, + }); + this.confirmed.set(partNumber, size); + } + + /** + * Commits the upload and waits for the storage backend to confirm, + * returning the blob's ETag or the reason the commit failed. The + * failure reason describes what went wrong but does not identify + * which parts, if any, were at fault; to retry, re-`putPart` (parts + * are safe to re-upload) and call `commit` again. + */ + async commit(options?: { signal?: AbortSignal }): Promise { + await this.call("Commit", {}); + while (true) { + options?.signal?.throwIfAborted(); + const info = await this.call("Info", {}); + if (info.status === "COMMITTED") { + return { etag: info.etag }; + } + if (info.commitError !== undefined && info.commitError !== "") { + return { error: info.commitError }; + } + await sleep(100); + } + } + + /** + * Uploads `data` in parts and commits: the whole story for bytes + * you already have. Resumes where a previous attempt left off. + */ + async upload( + data: globalThis.Blob | Uint8Array, + options?: UploadOptions + ): Promise { + // Refresh what the control plane already has, so interrupted + // uploads resume rather than restart. + const info = await this.call("Info", {}); + this.confirmed = new Map( + (info.parts ?? []).map( + (part: { number: number; size: string | number }) => [ + part.number, + Number(part.size), + ] + ) + ); + + const { partSize } = await this.instructions([], options); + const totalBytes = data instanceof Uint8Array ? data.byteLength : data.size; + const partCount = Math.max(1, Math.ceil(totalBytes / partSize)); + + let uploadedBytes = 0; + for (const [, size] of this.confirmed) { + uploadedBytes += size; + } + + for (let partNumber = 1; partNumber <= partCount; partNumber++) { + if (this.confirmed.has(partNumber)) { + continue; + } + const offset = (partNumber - 1) * partSize; + const bytes = data.slice(offset, Math.min(offset + partSize, totalBytes)); + await this.putPart(partNumber, bytes, options); + uploadedBytes += + bytes instanceof Uint8Array ? bytes.byteLength : bytes.size; + options?.onProgress?.({ uploadedBytes, totalBytes }); + } + + return await this.commit(options); + } +} + +/** + * The dead-simple upload hook. The blob id comes from an + * application-level RPC (blob creation is application-mediated), and + * then: + * + * const { upload } = useBlobUpload(); + * ... + * const { etag, error } = await upload(blobId, file); + */ +export function useBlobUpload(): { + upload: ( + blobId: string, + data: globalThis.Blob | Uint8Array, + options?: UploadOptions + ) => Promise; +} { + const client = useRebootClient(); + const upload = useMemo(() => { + return async ( + blobId: string, + data: globalThis.Blob | Uint8Array, + options?: UploadOptions + ) => { + const uploader = new BlobUploader({ + url: client.url, + blobId, + bearerToken: client.bearerToken, + }); + return await uploader.upload(data, options); + }; + }, [client.url, client.bearerToken]); + return { upload }; +} + +/** + * Resolves to a URL from which a committed blob's bytes can be + * downloaded (e.g. for an ``), or `undefined` while the blob + * is still uploading. Polls until the blob commits; render upload + * progress meanwhile via `useBlob(...).useInfo()`. + */ +export function useBlobDownloadUrl(blobId: string): string | undefined { + const client = useRebootClient(); + const [url, setUrl] = useState(undefined); + + useEffect(() => { + let unmounted = false; + const options = { + url: client.url, + blobId, + bearerToken: client.bearerToken, + }; + (async () => { + while (!unmounted) { + const info = await callBlobMethod(options, "Info", {}); + if (info.status === "COMMITTED") { + const response = await callBlobMethod(options, "DownloadUrl", {}); + if (!unmounted) { + setUrl(new URL(response.url, client.url).toString()); + } + return; + } + if (info.status === "DELETING" || info.status === "DELETED") { + return; + } + await sleep(500); + } + })().catch((error) => { + console.error(`[Reboot] Failed to resolve blob URL: ${error}`); + }); + return () => { + unmounted = true; + }; + }, [blobId, client.url, client.bearerToken]); + + return url; +} diff --git a/reboot/std/react/blobs/package.json b/reboot/std/react/blobs/package.json new file mode 100644 index 00000000..3dbc1ca5 --- /dev/null +++ b/reboot/std/react/blobs/package.json @@ -0,0 +1,3 @@ +{ + "type": "module" +} diff --git a/reboot/std/react/package.json b/reboot/std/react/package.json index 685c483f..9613c7f4 100644 --- a/reboot/std/react/package.json +++ b/reboot/std/react/package.json @@ -23,6 +23,7 @@ "exports": { "./package.json": "./package.json", ".": "./index.js", + "./blobs": "./blobs/index.js", "./presence": "./presence/index.js" } } diff --git a/tests/reboot/examples/chat-room/BUILD.bazel b/tests/reboot/examples/chat-room/BUILD.bazel index de5980d2..926f535d 100644 --- a/tests/reboot/examples/chat-room/BUILD.bazel +++ b/tests/reboot/examples/chat-room/BUILD.bazel @@ -124,9 +124,11 @@ sh_test_in_working_directory( # The locally built Reboot npm packages that the frontends are # type-checked against, instead of the published releases. frontend_packages = [ + "//rbt/std:reboot-dev-reboot-std-api", "//rbt/v1alpha1:reboot-dev-reboot-api", "//reboot:reboot.dev", "//reboot/nodejs:reboot-dev-reboot-" + REBOOT_VERSION + ".tgz", + "//reboot/std/react:reboot-dev-reboot-std-react", "//reboot/web:reboot-dev-reboot-web", "//reboot/react:reboot-dev-reboot-react", ] @@ -135,6 +137,8 @@ frontend_env = { "REBOOT_API_NPM_PACKAGE": "$(location //rbt/v1alpha1:reboot-dev-reboot-api)", "REBOOT_NPM_PACKAGE": "$(location //reboot/nodejs:reboot-dev-reboot-" + REBOOT_VERSION + ".tgz)", "REBOOT_REACT_NPM_PACKAGE": "$(location //reboot/react:reboot-dev-reboot-react)", + "REBOOT_STD_API_PACKAGE": "$(location //rbt/std:reboot-dev-reboot-std-api)", + "REBOOT_STD_REACT_PACKAGE": "$(location //reboot/std/react:reboot-dev-reboot-std-react)", "REBOOT_WEB_NPM_PACKAGE": "$(location //reboot/web:reboot-dev-reboot-web)", "REBOOT_WHL_FILE": "$(location //reboot:reboot.dev)", } diff --git a/tests/reboot/std/blobs/v1/BUILD.bazel b/tests/reboot/std/blobs/v1/BUILD.bazel new file mode 100644 index 00000000..7a5e1444 --- /dev/null +++ b/tests/reboot/std/blobs/v1/BUILD.bazel @@ -0,0 +1,16 @@ +load("@rbt_pypi//:requirements.bzl", "requirement") +load("@rules_python//python:defs.bzl", "py_test") + +py_test( + name = "blobs_tests_py", + size = "medium", + srcs = [":blobs_tests.py"], + main = "blobs_tests.py", + deps = [ + "//reboot/aio:applications_py", + "//reboot/aio:tests_py", + "//reboot/std/blobs/v1:blobs_py", + "@com_github_reboot_dev_reboot//rbt/std/blobs/v1:blobs_py_reboot", + requirement("aiohttp"), + ], +) diff --git a/tests/reboot/std/blobs/v1/blobs_tests.py b/tests/reboot/std/blobs/v1/blobs_tests.py new file mode 100644 index 00000000..b147d136 --- /dev/null +++ b/tests/reboot/std/blobs/v1/blobs_tests.py @@ -0,0 +1,477 @@ +import aiohttp +import asyncio +import hashlib +import tempfile +import unittest +from rbt.std.blobs.v1.blobs_rbt import ( + Blob, + Downloaders, + IncompleteParts, + InfoResponse, + NotCommitted, + SizeMismatch, +) +from reboot.aio.applications import Application +from reboot.aio.tests import Reboot +from reboot.std.blobs.v1._store import ( + DEFAULT_PART_SIZE_BYTES, + BlobStoreError, + FilesystemBlobStore, + UploadedPart, + _encode_blob_id, +) +from reboot.std.blobs.v1.blobs import blobs_library + + +class TestBlobs(unittest.IsolatedAsyncioTestCase): + """Exercises the `Blob` control plane against the filesystem data + plane that the `Reboot` test harness runs, through the same + reverse-proxied byte routes an application serves under + `rbt dev run`.""" + + async def asyncSetUp(self) -> None: + self.rbt = Reboot() + await self.rbt.start() + + await self.rbt.up( + Application(libraries=[blobs_library()]), + local_envoy=True, + ) + + self.context = self.rbt.create_external_context( + name=f"test-{self.id()}", + app_internal=True, + ) + self.external_context = self.rbt.create_external_context( + name=f"test-external-{self.id()}", + ) + + async def asyncTearDown(self) -> None: + await self.rbt.stop() + + async def _instructions(self, blob, part_numbers: list[int]): + """Fetches upload instructions, waiting for the `BeginUpload` + workflow to have provisioned the upload session.""" + while True: + response = await blob.upload_instructions( + self.context, + part_numbers=part_numbers, + ) + if response.ready: + return response + await asyncio.sleep(0.05) + + async def _put(self, url: str, data: bytes) -> str: + """`PUT`s bytes to a (possibly relative) data-plane URL and + returns the response's ETag.""" + async with aiohttp.ClientSession(self.rbt.url()) as session: + async with session.put(url, data=data) as response: + if response.status != 200: + self.fail( + f"PUT failed ({response.status}): " + f"{await response.text()}" + ) + return response.headers["ETag"].strip('"') + + async def _upload(self, blob, data: bytes) -> None: + """Uploads `data` in data-plane-sized parts and reports each + part, exactly as the browser SDK does.""" + part_size = (await self._instructions(blob, [])).part_size + parts = [ + data[offset:offset + part_size] + for offset in range(0, len(data), part_size) + ] or [b""] + instructions = await self._instructions( + blob, list(range(1, + len(parts) + 1)) + ) + for instruction, part in zip(instructions.instructions, parts): + etag = await self._put(instruction.url, part) + await blob.part_uploaded( + self.context, + part_number=instruction.part_number, + etag=etag, + size=len(part), + ) + + async def _wait_until_status(self, blob, statuses) -> InfoResponse: + while True: + info = await blob.info(self.context) + if info.status in statuses: + return info + await asyncio.sleep(0.05) + + async def _download(self, blob) -> tuple[bytes, str]: + """Downloads the blob's bytes via its download URL, returning + the bytes and the response's content type.""" + url = (await blob.download_url(self.context)).url + async with aiohttp.ClientSession(self.rbt.url()) as session: + async with session.get(url) as response: + if response.status != 200: + self.fail( + f"GET failed ({response.status}): " + f"{await response.text()}" + ) + return await response.read(), response.content_type + + async def test_upload_and_download(self) -> None: + # 2.5 data-plane parts, so the upload is genuinely multipart. + data = bytes(range(256)) * (DEFAULT_PART_SIZE_BYTES * 5 // 2 // 256) + + blob, _ = await Blob.create( + self.context, + content_type="application/octet-stream", + size=len(data), + ) + + await self._upload(blob, data) + + info = await blob.info(self.context) + self.assertEqual(info.bytes_uploaded, len(data)) + self.assertEqual(info.status, Blob.State.UPLOADING) + + await blob.commit(self.context) + + info = await self._wait_until_status(blob, {Blob.State.COMMITTED}) + self.assertTrue(info.etag.endswith("-3")) + + downloaded, content_type = await self._download(blob) + self.assertEqual(downloaded, data) + self.assertEqual(content_type, "application/octet-stream") + + async def test_commit_validates_etags(self) -> None: + data = b"hello, blobs!" + + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + + instructions = await self._instructions(blob, [1]) + await self._put(instructions.instructions[0].url, data) + + # Report a bogus ETag; the commit must fail and revert the + # blob to `UPLOADING` with `commit_error` set. + await blob.part_uploaded( + self.context, + part_number=1, + etag=hashlib.md5(b"not the data").hexdigest(), + size=len(data), + ) + await blob.commit(self.context) + + info = await self._wait_until_status(blob, {Blob.State.UPLOADING}) + self.assertIn("ETag mismatch", info.commit_error) + + # Re-report the correct ETag and commit again; now it must + # succeed. + await blob.part_uploaded( + self.context, + part_number=1, + etag=hashlib.md5(data).hexdigest(), + size=len(data), + ) + await blob.commit(self.context) + info = await self._wait_until_status(blob, {Blob.State.COMMITTED}) + self.assertFalse(info.HasField("commit_error")) + + downloaded, _ = await self._download(blob) + self.assertEqual(downloaded, data) + + async def test_size_validation(self) -> None: + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + max_size=10, + ) + + with self.assertRaises(Blob.PartUploadedAborted) as raised: + await blob.part_uploaded( + self.context, + part_number=1, + etag="0" * 32, + size=100, + ) + self.assertIsInstance(raised.exception.error, SizeMismatch) + + # Declared `size` must match the sum of the parts at commit + # time. + sized_blob, _ = await Blob.create( + self.context, + content_type="text/plain", + size=5, + ) + await sized_blob.part_uploaded( + self.context, + part_number=1, + etag="0" * 32, + size=3, + ) + with self.assertRaises(Blob.CommitAborted) as commit_raised: + await sized_blob.commit(self.context) + self.assertIsInstance(commit_raised.exception.error, SizeMismatch) + + async def test_commit_requires_contiguous_parts(self) -> None: + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + await blob.part_uploaded( + self.context, + part_number=2, + etag="0" * 32, + size=1, + ) + with self.assertRaises(Blob.CommitAborted) as raised: + await blob.commit(self.context) + self.assertIsInstance(raised.exception.error, IncompleteParts) + + async def test_download_url_requires_committed(self) -> None: + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + with self.assertRaises(Blob.DownloadUrlAborted) as raised: + await blob.download_url(self.context) + self.assertIsInstance(raised.exception.error, NotCommitted) + + async def test_store_complete_enforces_max_size(self) -> None: + # `complete()` verifies the *real* total size against + # `max_size` independently of the control plane's own + # (reported-size) check, as defense in depth against untruthful + # reported sizes. A store-level test, on its own store. + with tempfile.TemporaryDirectory() as directory: + store = FilesystemBlobStore(directory) + blob_id = "max-size-blob" + upload_id = await store.begin_upload(blob_id, "text/plain") + encoded = _encode_blob_id(blob_id) + data = b"x" * 100 + with open(store.part_path(encoded, upload_id, 1), "wb") as f: + f.write(data) + part = UploadedPart( + number=1, + etag=hashlib.md5(data).hexdigest(), + size=len(data), + ) + with self.assertRaises(BlobStoreError): + await store.complete( + blob_id, + upload_id, + "text/plain", + [part], + max_size=50, + ) + # Within the bound, it succeeds. + etag = await store.complete( + blob_id, + upload_id, + "text/plain", + [part], + max_size=1000, + ) + self.assertTrue(etag.endswith("-1")) + + async def test_commit_rejects_zero_parts(self) -> None: + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + with self.assertRaises(Blob.CommitAborted) as raised: + await blob.commit(self.context) + self.assertIsInstance(raised.exception.error, IncompleteParts) + + async def test_part_uploaded_rejects_bad_input(self) -> None: + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + # Part number 0 (e.g. an omitted proto field) is rejected. + with self.assertRaises(Blob.PartUploadedAborted): + await blob.part_uploaded( + self.context, + part_number=0, + etag="0" * 32, + size=1, + ) + # A non-MD5-hex ETag is rejected (it would corrupt the S3 + # completion XML). + with self.assertRaises(Blob.PartUploadedAborted): + await blob.part_uploaded( + self.context, + part_number=1, + etag='">', + size=1, + ) + + async def test_uploader_gating(self) -> None: + # A blob with an uploader: external callers that are not the + # uploader may read `Info` but not upload. (This application + # has no token verifier, so the external context has no user + # at all.) + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + uploader_id="alice", + ) + + info = await Blob.ref(blob.state_id).info(self.external_context) + self.assertEqual(info.uploader_id, "alice") + + # A non-uploader (here: an unauthenticated external caller) is + # denied upload-side calls. Authorization denials surface as + # the method's `Aborted` type. + with self.assertRaises(Blob.PartUploadedAborted): + await Blob.ref(blob.state_id).part_uploaded( + self.external_context, + part_number=1, + etag="0" * 32, + size=1, + ) + + # A blob without an uploader: anyone who knows the id may + # upload. Use a fresh external context: after the denied + # mutation above, the previous context considers the outcome + # of its last mutation uncertain and refuses further + # non-idempotent mutations. + open_blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + open_context = self.rbt.create_external_context( + name=f"test-open-{self.id()}", + ) + await Blob.ref(open_blob.state_id).part_uploaded( + open_context, + part_number=1, + etag="0" * 32, + size=1, + ) + + async def test_download_gating(self) -> None: + data = b"secret bytes" + + # A blob with a download allow-list: only listed users may get + # a download URL. This application has no token verifier, so + # the external context has no user and is not on any list. + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + downloader_ids=Downloaders(user_ids=["bob"]), + ) + await self._upload(blob, data) + await blob.commit(self.context) + await self._wait_until_status(blob, {Blob.State.COMMITTED}) + + # The app-internal caller may always download. + downloaded, _ = await self._download(blob) + self.assertEqual(downloaded, data) + + # A caller not on the allow-list is denied. + with self.assertRaises(Blob.DownloadUrlAborted): + await Blob.ref(blob.state_id).download_url(self.external_context) + + # A blob with no allow-list: anyone who knows the id may get a + # download URL. + open_blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + await self._upload(open_blob, data) + await open_blob.commit(self.context) + await self._wait_until_status(open_blob, {Blob.State.COMMITTED}) + + response = await Blob.ref(open_blob.state_id + ).download_url(self.external_context) + self.assertNotEqual(response.url, "") + + async def test_set_downloaders(self) -> None: + data = b"mutable acl" + + # Start open: anyone who knows the id may download. + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + await self._upload(blob, data) + await blob.commit(self.context) + await self._wait_until_status(blob, {Blob.State.COMMITTED}) + + response = await Blob.ref(blob.state_id + ).download_url(self.external_context) + self.assertNotEqual(response.url, "") + + # Restrict downloads to a user the external caller is not. + await blob.set_downloaders( + self.context, + downloader_ids=Downloaders(user_ids=["bob"]), + ) + with self.assertRaises(Blob.DownloadUrlAborted): + await Blob.ref(blob.state_id).download_url(self.external_context) + + # Remove the restriction again by omitting `downloader_ids`. + await blob.set_downloaders(self.context) + response = await Blob.ref(blob.state_id + ).download_url(self.external_context) + self.assertNotEqual(response.url, "") + + async def test_info_gating(self) -> None: + # `Info` is visible to anyone who may upload or download the + # blob. With both sides restricted, an unauthenticated external + # caller (this app has no token verifier) can do neither, so it + # cannot read `Info` either. + locked, _ = await Blob.create( + self.context, + content_type="text/plain", + uploader_id="alice", + downloader_ids=Downloaders(user_ids=["bob"]), + ) + with self.assertRaises(Blob.InfoAborted): + await Blob.ref(locked.state_id).info(self.external_context) + + # Open upload side (empty `uploader_id`): anyone who may upload + # may also watch progress via `Info`, even behind a download + # allow-list. + upload_open, _ = await Blob.create( + self.context, + content_type="text/plain", + downloader_ids=Downloaders(user_ids=["bob"]), + ) + info = await Blob.ref(upload_open.state_id).info(self.external_context) + self.assertEqual(info.status, Blob.State.UPLOADING) + + # Open download side (omitted `downloader_ids`): anyone who may + # download may read `Info`, even with a specific uploader. + download_open, _ = await Blob.create( + self.context, + content_type="text/plain", + uploader_id="alice", + ) + info = await Blob.ref(download_open.state_id + ).info(self.external_context) + self.assertEqual(info.status, Blob.State.UPLOADING) + + async def test_delete(self) -> None: + data = b"delete me" + + blob, _ = await Blob.create( + self.context, + content_type="text/plain", + ) + await self._upload(blob, data) + await blob.commit(self.context) + await self._wait_until_status(blob, {Blob.State.COMMITTED}) + + url = (await blob.download_url(self.context)).url + + await blob.remove(self.context) + info = await self._wait_until_status(blob, {Blob.State.DELETED}) + self.assertEqual(info.status, Blob.State.DELETED) + + # The bytes must be gone from the data plane. + async with aiohttp.ClientSession(self.rbt.url()) as session: + async with session.get(url) as response: + self.assertEqual(response.status, 404) + + +if __name__ == "__main__": + unittest.main()