more invariant fixes (#306)

## Why + What changed

- fix `got stream cancel without a valid protocol error`
- fix `session state has been consumed and is no longer valid: getting
loggingMetadata on consumed state` (logging after protocol error +
session deletion)

## Versioning

- [ ] Breaking protocol change
- [ ] Breaking ts/js API change

<!-- Kind reminder to add tests and updated documentation if needed -->

Author: jackyzha0

router/client.ts

@@ -32,14 +32,11 @@ import { Value } from '@sinclair/typebox/value';
 import { PayloadType, ValidProcType } from './procedures';
 import {
   BaseErrorSchemaType,
-  ErrResultSchema,
   CANCEL_CODE,
-  ReaderErrorSchema,
+  ReaderErrorResultSchema,
   UNEXPECTED_DISCONNECT_CODE,
 } from './errors';
 
-const ReaderErrResultSchema = ErrResultSchema(ReaderErrorSchema);
-
 interface CallOptions {
   signal?: AbortSignal;
 }
@@ -384,22 +381,22 @@ function handleProc(
       cleanClose = false;
 
       span.addEvent('received cancel');
-      let cancelResult: Static<typeof ReaderErrResultSchema>;
+      let cancelResult: Static<typeof ReaderErrorResultSchema>;
 
-      if (Value.Check(ReaderErrResultSchema, msg.payload)) {
+      if (Value.Check(ReaderErrorResultSchema, msg.payload)) {
         cancelResult = msg.payload;
       } else {
         cancelResult = Err({
           code: CANCEL_CODE,
           message: 'stream cancelled with invalid payload',
         });
-        transport.log?.error(
+        transport.log?.warn(
           'got stream cancel without a valid protocol error',
           {
             clientId: transport.clientId,
             transportMessage: msg,
             validationErrors: [
-              ...Value.Errors(ReaderErrResultSchema, msg.payload),
+              ...Value.Errors(ReaderErrorResultSchema, msg.payload),
             ],
           },
         );

router/context.ts

@@ -2,7 +2,7 @@ import { Span } from '@opentelemetry/api';
 import { TransportClientId } from '../transport/message';
 import { SessionId } from '../transport/sessionStateMachine/common';
 import { ErrResult } from './result';
-import { CancelResultSchema } from './errors';
+import { CancelErrorSchema } from './errors';
 import { Static } from '@sinclair/typebox';
 
 /**
@@ -78,7 +78,7 @@ export type ProcedureHandlerContext<State> = ServiceContext & {
    * Cancelling is not the same as closing procedure calls gracefully, please refer to
    * the river documentation to understand the difference between the two concepts.
    */
-  cancel: (message?: string) => ErrResult<Static<typeof CancelResultSchema>>;
+  cancel: (message?: string) => ErrResult<Static<typeof CancelErrorSchema>>;
   /**
    * This signal is a standard [AbortSignal](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal)
    * triggered when the procedure invocation is done. This signal tracks the invocation/request finishing

router/errors.ts

@@ -75,11 +75,13 @@ export function castTypeboxValueErrors(
 /**
  * A schema for cancel payloads sent from the client
  */
-export const CancelResultSchema = Type.Object({
+export const CancelErrorSchema = Type.Object({
   code: Type.Literal(CANCEL_CODE),
   message: Type.String(),
 });
 
+export const CancelResultSchema = ErrResultSchema(CancelErrorSchema);
+
 /**
  * {@link ReaderErrorSchema} is the schema for all the built-in river errors that
  * can be emitted to a reader (request reader on the server, and response reader
@@ -104,9 +106,11 @@ export const ReaderErrorSchema = Type.Union([
       }),
     ),
   }),
-  CancelResultSchema,
+  CancelErrorSchema,
 ]) satisfies ProcedureErrorSchemaType;
 
+export const ReaderErrorResultSchema = ErrResultSchema(ReaderErrorSchema);
+
 /**
  * Represents an acceptable schema to pass to a procedure.
  * Just a type of a schema, not an actual schema.

router/procedures.ts

@@ -4,7 +4,7 @@ import { ProcedureHandlerContext } from './context';
 import { Result } from './result';
 import { Readable, Writable } from './streams';
 import {
-  CancelResultSchema,
+  CancelErrorSchema,
   ProcedureErrorSchemaType,
   ReaderErrorSchema,
 } from './errors';
@@ -39,7 +39,7 @@ export type ValidProcType =
  */
 export type PayloadType = TSchema;
 
-export type Cancellable<T> = T | Static<typeof CancelResultSchema>;
+export type Cancellable<T> = T | Static<typeof CancelErrorSchema>;
 
 /**
  * Procedure for a single message in both directions (1:1).

router/server.ts

@@ -283,9 +283,9 @@ class RiverServer<Services extends AnyServiceSchemaMap>
       }
 
       if (isStreamCancelBackwardsCompat(msg.controlFlags, protocolVersion)) {
-        let cancelResult: ErrResult<Static<typeof CancelResultSchema>>;
+        let cancelResult: Static<typeof CancelResultSchema>;
         if (Value.Check(CancelResultSchema, msg.payload)) {
-          cancelResult = Err(msg.payload);
+          cancelResult = msg.payload;
         } else {
           // If the payload is unexpected, then we just construct our own cancel result
           cancelResult = Err({

transport/client.ts

@@ -319,17 +319,6 @@ export abstract class ClientTransport<
 
     const res = connectedSession.sendBufferedMessages();
     if (!res.ok) {
-      this.log?.error(`failed to send buffered messages: ${res.reason}`, {
-        ...connectedSession.loggingMetadata,
-        transportMessage: msg,
-      });
-
-      this.protocolError({
-        type: ProtocolError.MessageSendFailure,
-        message: res.reason,
-      });
-      this.deleteSession(connectedSession, { unhealthy: true });
-
       return;
     }
 

transport/server.ts

@@ -545,20 +545,6 @@ export abstract class ServerTransport<
 
     const bufferSendRes = connectedSession.sendBufferedMessages();
     if (!bufferSendRes.ok) {
-      this.log?.error(
-        `failed to send buffered messages: ${bufferSendRes.reason}`,
-        {
-          ...connectedSession.loggingMetadata,
-          transportMessage: msg,
-        },
-      );
-
-      this.protocolError({
-        type: ProtocolError.MessageSendFailure,
-        message: bufferSendRes.reason,
-      });
-      this.deleteSession(connectedSession, { unhealthy: true });
-
       return;
     }
 

transport/transport.ts

@@ -260,7 +260,7 @@ export abstract class Transport<ConnType extends Connection> {
 
   // common listeners
   protected onSessionGracePeriodElapsed(session: Session<ConnType>) {
-    this.log?.warn(
+    this.log?.info(
       `session to ${session.to} grace period elapsed, closing`,
       session.loggingMetadata,
     );