Server uses ReadStream and WriteStream #136 #137

masad-frost · 2024-05-21T23:20:40Z

Exactly the same as #136 but for the server

masad-frost · 2024-05-21T23:23:13Z

router/server.ts

+    const hasInit = 'init' in procedure;
+
+    if (!isFirstMessage || !hasInit) {
+      // Init message is consumed during stream instantiation
+      if (Value.Check(procedure.input, message.payload)) {
+        procStream.incoming.pushValue(message.payload as PayloadType);
+      } else if (!Value.Check(ControlMessagePayloadSchema, message.payload)) {
+        log?.error(
+          `procedure ${serviceName}.${procedureName} received invalid payload`,
+          { clientId: this.transport.clientId, fullTransportMessage: message },
+        );
+      }


I don't like this approach (or any of the changes i made to this file for that matter.) There's going to have to be some refactors for us to implement invalid requests, half-close properly, and close requests, so I'm gonna stick to keeping changes to here as minimal as possible.

yeah this is pretty hard to parse in some points, it feels inconsistent how and when we check init vs payload at least within the diff

masad-frost · 2024-05-21T23:24:12Z

router/server.ts

+    const hasInit = 'init' in procedure;
+
+    if (!isFirstMessage || !hasInit) {


in theory this shouldn't work for rpc, but it does 🤷

We just end up pushing extraneous values to the stream.

what do we end up pushing 🤔

masad-frost · 2024-05-21T23:26:27Z

router/server.ts

+    if (!stream.incoming.isClosed()) {
+      stream.incoming.triggerClose();
+    }


should probably call drain here. Also don't need to be checking isClosed, but it's ok for now

jackyzha0 · 2024-05-23T05:37:27Z

router/server.ts

@@ -326,12 +302,24 @@ class RiverServer<Services extends AnyServiceSchemaMap> {
            },
            activeContext,
            async (span: Span) => {
+              if (!Value.Check(procedure.input, message.payload)) {


shouldn't the first (and only message) on a subscription to the server be init? im a little confused, is this + #136 rewriting to also do the thing we talked about where all streams are mandatory stream init message + potentially more body messages instead of the current state of the world where it is optional special init message + maybe more body messages

This is the RPC handler btw, first message don't go into the stream.

Yeah if we have mandatory inits the code will be much cleaner

subscriptions and rpc currently are not init. I'll update the protocol doc then readapt the changes

jackyzha0 · 2024-05-23T05:38:07Z

router/server.ts

+    const hasInit = 'init' in procedure;
+
+    if (!isFirstMessage || !hasInit) {


what do we end up pushing 🤔

jackyzha0 · 2024-05-23T05:40:40Z

router/server.ts

+    const hasInit = 'init' in procedure;
+
+    if (!isFirstMessage || !hasInit) {
+      // Init message is consumed during stream instantiation
+      if (Value.Check(procedure.input, message.payload)) {
+        procStream.incoming.pushValue(message.payload as PayloadType);
+      } else if (!Value.Check(ControlMessagePayloadSchema, message.payload)) {
+        log?.error(
+          `procedure ${serviceName}.${procedureName} received invalid payload`,
+          { clientId: this.transport.clientId, fullTransportMessage: message },
+        );
+      }


yeah this is pretty hard to parse in some points, it feels inconsistent how and when we check init vs payload at least within the diff

jackyzha0 · 2024-05-23T05:41:34Z

util/testHelpers.ts

+  const inputPipe = createPipe<Static<I>>();
+  const outputPipe = createPipe<ProcedureResult<O, E>>();


ooo this is nice

masad-frost · 2024-05-28T21:45:23Z

router/server.ts

@@ -448,10 +444,22 @@ class RiverServer<Services extends AnyServiceSchemaMap> {
            },
            activeContext,
            async (span: Span) => {
+              if (!Value.Check(procedure.input, message.payload)) {


yeah this should check init :/

or wait, subscriptions don't have init, only input.

* Server uses streams * Bruh

All the changes are documented in `Protocol.md` but here's a summary: - Handle invalid client requests by sending a close with an error back - This was the main motivation for the change. While we could sort-of implement this error response without the other changes, things are setup in such a way where it is very hard to implement correctly without deeper changes in how we handle closing. - Add more robust closing mechanics - Half-close states - Close signals from read end of the pipes - Abort full-closure (for errors and cancellation) - Switch from `Pushable` and `AsyncIterator` APIs to a `ReadStream` and `WriteStream` - All procedures have `init` and some have `input` While the changes are not strictly backwards compatible, hence the major protocol bump, the system can still operate across versions to some extent. See PRs linked below for more information on the above # TODOs - [x] Define protocol and update doc #111 - [x] Design stream abstractions #118 - [x] Redsigned in #249 - [x] Implement stream abstractions - [x] ReadStream #130 - [x] WriteStream #132 - [x] All streams have init, some have input. - [x] Protocol change documented in #153 - [x] Implementation change #159 - [x] Use stream abstractions & implement protocol closing semantics - [x] Protocol: Implement close requests from readers #165 - [x] Protocol: Implement half-close - [x] Client #162 - [x] Server #163 - [x] Simple s/Pushable/Stream replacement - [x] Client #136 - [x] Server #137 - [x] Make `Input` iterator on the server use `Result` so we can signal stream closes, client disconnects, and aborts #172 - [x] Add Abort mechanism - [x] Docs update #175 - [x] Implement abort - [x] Client #193 - [x] Server #200 - [x] Add `INVALID_REQUEST` to schema #107 - [x] Handle/send back `INVALID_REQUEST` errors with an abort bit #203 - [x] Handle/send back `INTERNAL_RIVER_ERROR` with an abort bit #203 - [x] Send abort bit with `UNCAUGHT_ERROR` #201 - [x] Abort tombstones #204 - [ ] Try to find uncovered areas to test - [ ] `undefined` value for `init`, `input`, & `output`. - [ ] Update docs - [ ] Changelog --------- Co-authored-by: Jacky Zhao <j.zhao2k19@gmail.com>

masad-frost requested a review from a team as a code owner May 21, 2024 23:20

masad-frost requested review from jackyzha0 and removed request for a team May 21, 2024 23:20

masad-frost commented May 21, 2024

View reviewed changes

Base automatically changed from fm-client-use-streams to protocolv2 May 21, 2024 23:33

masad-frost force-pushed the fm-server-use-streams branch from 82d0a80 to cb18527 Compare May 21, 2024 23:36

masad-frost mentioned this pull request May 21, 2024

Base branch for protocol v2 #127

Merged

32 tasks

masad-frost force-pushed the protocolv2 branch from 9be4fa0 to ad39e56 Compare May 21, 2024 23:39

masad-frost added 2 commits May 21, 2024 23:56

Server uses streams

373393a

Bruh

9f7d8a9

masad-frost force-pushed the fm-server-use-streams branch from cb18527 to 9f7d8a9 Compare May 21, 2024 23:56

jackyzha0 approved these changes May 23, 2024

View reviewed changes

masad-frost commented May 28, 2024

View reviewed changes

masad-frost merged commit 45072e5 into protocolv2 May 28, 2024
3 checks passed

masad-frost deleted the fm-server-use-streams branch May 28, 2024 21:52

masad-frost added a commit that referenced this pull request May 29, 2024

Server uses ReadStream and WriteStream #136 (#137)

be75882

* Server uses streams * Bruh

masad-frost added a commit that referenced this pull request May 31, 2024

Server uses ReadStream and WriteStream #136 (#137)

682adad

* Server uses streams * Bruh

masad-frost added a commit that referenced this pull request Jun 6, 2024

Server uses ReadStream and WriteStream #136 (#137)

87b15ec

* Server uses streams * Bruh

masad-frost added a commit that referenced this pull request Jun 11, 2024

Server uses ReadStream and WriteStream #136 (#137)

2143a34

* Server uses streams * Bruh

masad-frost added a commit that referenced this pull request Jun 13, 2024

Server uses ReadStream and WriteStream #136 (#137)

c250646

* Server uses streams * Bruh

masad-frost added a commit that referenced this pull request Jun 20, 2024

Server uses ReadStream and WriteStream #136 (#137)

7a92a81

* Server uses streams * Bruh

masad-frost added a commit that referenced this pull request Jun 24, 2024

Server uses ReadStream and WriteStream #136 (#137)

4542ee1

* Server uses streams * Bruh

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Server uses ReadStream and WriteStream #136 #137

Server uses ReadStream and WriteStream #136 #137

masad-frost commented May 21, 2024

masad-frost May 21, 2024

jackyzha0 May 23, 2024

masad-frost May 21, 2024

masad-frost May 21, 2024 •

edited

Loading

jackyzha0 May 23, 2024

masad-frost May 21, 2024

jackyzha0 May 23, 2024

masad-frost May 28, 2024

masad-frost May 28, 2024

jackyzha0 May 23, 2024

jackyzha0 May 23, 2024

jackyzha0 May 23, 2024

masad-frost May 28, 2024

masad-frost May 28, 2024

		const hasInit = 'init' in procedure;

		if (!isFirstMessage \|\| !hasInit) {

		const inputPipe = createPipe<Static<I>>();
		const outputPipe = createPipe<ProcedureResult<O, E>>();

Server uses ReadStream and WriteStream #136 #137

Server uses ReadStream and WriteStream #136 #137

Conversation

masad-frost commented May 21, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

masad-frost May 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

masad-frost May 21, 2024 •

edited

Loading