nnrp-core 1.0.0-preview.4.8

Canonical NNRP wire codecs, protocol validation, state machines, and preview3 core types.
Documentation
use crate::{
    NnrpError, SessionMigrateAckMetadata, SessionMigrateMetadata, SessionOpenAckMetadata,
    SessionOpenMetadata, SessionStatus, SESSION_ERROR_NONE, SESSION_ERROR_RESUME_REJECTED,
};

pub const SESSION_FLAG_ALLOW_RESUME: u8 = 0x01;
pub const SESSION_ACK_FLAG_RESUME_ENABLED: u32 = 0x0000_0001;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct SessionRecoveryIntent {
    pub requested_session_id: u32,
    pub resume_token_bytes: u32,
    pub resume_from_operation_id: Option<u64>,
}

impl SessionRecoveryIntent {
    pub fn from_session_open(
        metadata: &SessionOpenMetadata,
        resume_from_operation_id: Option<u64>,
    ) -> Self {
        Self {
            requested_session_id: metadata.requested_session_id,
            resume_token_bytes: metadata.resume_token_bytes,
            resume_from_operation_id,
        }
    }

    pub fn is_resume_attempt(&self) -> bool {
        self.resume_token_bytes > 0
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SessionRecoveryOutcome {
    Fresh,
    ResumeEnabled { resume_window_ms: u32 },
    Resumed { resume_window_ms: u32 },
    ResumeRejected,
}

pub fn validate_session_recovery_request(metadata: &SessionOpenMetadata) -> Result<(), NnrpError> {
    if metadata.resume_token_bytes > 0 {
        if metadata.session_flags & SESSION_FLAG_ALLOW_RESUME == 0 {
            return Err(NnrpError::InvalidProtocolCombination {
                rule: "resume token requires allow_resume session flag",
            });
        }

        if metadata.requested_session_id == 0 {
            return Err(NnrpError::InvalidProtocolCombination {
                rule: "resume token must be bound to a requested session id",
            });
        }
    }

    Ok(())
}

pub fn validate_session_recovery_ack(
    request: &SessionOpenMetadata,
    ack: &SessionOpenAckMetadata,
) -> Result<SessionRecoveryOutcome, NnrpError> {
    let resume_enabled = ack.session_flags_ack & SESSION_ACK_FLAG_RESUME_ENABLED != 0;

    if resume_enabled && (ack.resume_window_ms == 0 || ack.resume_token_bytes == 0) {
        return Err(NnrpError::InvalidProtocolCombination {
            rule: "resume_enabled ack requires non-zero resume window and token bytes",
        });
    }

    match ack.session_status {
        SessionStatus::Opened => {
            if resume_enabled {
                Ok(SessionRecoveryOutcome::ResumeEnabled {
                    resume_window_ms: ack.resume_window_ms,
                })
            } else {
                Ok(SessionRecoveryOutcome::Fresh)
            }
        }
        SessionStatus::Resumed => {
            if request.resume_token_bytes == 0 {
                return Err(NnrpError::InvalidProtocolCombination {
                    rule: "resumed ack requires a resume token in the request",
                });
            }
            if ack.session_error_code != SESSION_ERROR_NONE {
                return Err(NnrpError::InvalidProtocolCombination {
                    rule: "resumed ack must not carry a session error",
                });
            }
            if !resume_enabled {
                return Err(NnrpError::InvalidProtocolCombination {
                    rule: "resumed ack must keep resume_enabled set",
                });
            }

            Ok(SessionRecoveryOutcome::Resumed {
                resume_window_ms: ack.resume_window_ms,
            })
        }
        SessionStatus::Rejected if ack.session_error_code == SESSION_ERROR_RESUME_REJECTED => {
            Ok(SessionRecoveryOutcome::ResumeRejected)
        }
        SessionStatus::Rejected | SessionStatus::RetryLater => Ok(SessionRecoveryOutcome::Fresh),
    }
}

pub fn validate_migration_recovery(
    request: &SessionMigrateMetadata,
    ack: &SessionMigrateAckMetadata,
) -> Result<(), NnrpError> {
    if request.old_transport_id == request.new_transport_id {
        return Err(NnrpError::InvalidProtocolCombination {
            rule: "session migration must bind a different transport",
        });
    }

    if ack.resume_from_frame_id < request.last_result_frame_id {
        return Err(NnrpError::InvalidProtocolCombination {
            rule: "migration resume cursor cannot move behind the last result frame",
        });
    }

    Ok(())
}

pub fn should_replay_frame_after_migration(ack: &SessionMigrateAckMetadata, frame_id: u64) -> bool {
    frame_id >= ack.resume_from_frame_id
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::TransportId;

    #[test]
    fn resume_request_requires_flag_and_session_binding() {
        let mut request = session_open(42, 16, 0);

        assert_eq!(
            validate_session_recovery_request(&request),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "resume token requires allow_resume session flag"
            })
        );

        request.session_flags = SESSION_FLAG_ALLOW_RESUME;
        request.requested_session_id = 0;
        assert_eq!(
            validate_session_recovery_request(&request),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "resume token must be bound to a requested session id"
            })
        );

        request.requested_session_id = 42;
        assert_eq!(validate_session_recovery_request(&request), Ok(()));
    }

    #[test]
    fn recovery_ack_exports_fresh_enabled_resumed_and_rejected_outcomes() {
        let request = session_open(42, 16, SESSION_FLAG_ALLOW_RESUME);

        let enabled = session_ack(
            42,
            SessionStatus::Opened,
            SESSION_ACK_FLAG_RESUME_ENABLED,
            10_000,
            24,
            SESSION_ERROR_NONE,
        );
        assert_eq!(
            validate_session_recovery_ack(&request, &enabled),
            Ok(SessionRecoveryOutcome::ResumeEnabled {
                resume_window_ms: 10_000
            })
        );

        let resumed = session_ack(
            42,
            SessionStatus::Resumed,
            SESSION_ACK_FLAG_RESUME_ENABLED,
            10_000,
            24,
            SESSION_ERROR_NONE,
        );
        assert_eq!(
            validate_session_recovery_ack(&request, &resumed),
            Ok(SessionRecoveryOutcome::Resumed {
                resume_window_ms: 10_000
            })
        );

        let rejected = session_ack(
            0,
            SessionStatus::Rejected,
            0,
            0,
            0,
            SESSION_ERROR_RESUME_REJECTED,
        );
        assert_eq!(
            validate_session_recovery_ack(&request, &rejected),
            Ok(SessionRecoveryOutcome::ResumeRejected)
        );

        let fresh = session_ack(43, SessionStatus::Opened, 0, 0, 0, SESSION_ERROR_NONE);
        assert_eq!(
            validate_session_recovery_ack(&session_open(0, 0, 0), &fresh),
            Ok(SessionRecoveryOutcome::Fresh)
        );
    }

    #[test]
    fn recovery_ack_rejects_incomplete_resumed_state() {
        let request = session_open(42, 16, SESSION_FLAG_ALLOW_RESUME);
        let mut ack = session_ack(42, SessionStatus::Resumed, 0, 0, 0, SESSION_ERROR_NONE);

        assert_eq!(
            validate_session_recovery_ack(&request, &ack),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "resumed ack must keep resume_enabled set"
            })
        );

        ack.session_flags_ack = SESSION_ACK_FLAG_RESUME_ENABLED;
        ack.resume_window_ms = 10_000;
        ack.resume_token_bytes = 24;
        ack.session_error_code = SESSION_ERROR_RESUME_REJECTED;
        assert_eq!(
            validate_session_recovery_ack(&request, &ack),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "resumed ack must not carry a session error"
            })
        );

        ack.session_error_code = SESSION_ERROR_NONE;
        assert_eq!(
            validate_session_recovery_ack(&session_open(42, 0, SESSION_FLAG_ALLOW_RESUME), &ack),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "resumed ack requires a resume token in the request"
            })
        );
    }

    #[test]
    fn migration_recovery_consumes_resume_cursor() {
        let request = SessionMigrateMetadata {
            old_transport_id: TransportId::Tcp,
            new_transport_id: TransportId::Quic,
            last_result_frame_id: 10,
            client_migrate_ts_us: 100,
        };
        let ack = SessionMigrateAckMetadata {
            accept_code: 0,
            resume_from_frame_id: 12,
            grace_window_ms: 500,
            server_migrate_ts_us: 200,
        };

        assert_eq!(validate_migration_recovery(&request, &ack), Ok(()));
        assert!(!should_replay_frame_after_migration(&ack, 11));
        assert!(should_replay_frame_after_migration(&ack, 12));

        let stale_ack = SessionMigrateAckMetadata {
            resume_from_frame_id: 9,
            ..ack
        };
        assert_eq!(
            validate_migration_recovery(&request, &stale_ack),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "migration resume cursor cannot move behind the last result frame"
            })
        );
    }

    #[test]
    fn migration_requires_distinct_bindings() {
        let request = SessionMigrateMetadata {
            old_transport_id: TransportId::Quic,
            new_transport_id: TransportId::Quic,
            last_result_frame_id: 1,
            client_migrate_ts_us: 100,
        };
        let ack = SessionMigrateAckMetadata {
            accept_code: 0,
            resume_from_frame_id: 1,
            grace_window_ms: 500,
            server_migrate_ts_us: 200,
        };

        assert_eq!(
            validate_migration_recovery(&request, &ack),
            Err(NnrpError::InvalidProtocolCombination {
                rule: "session migration must bind a different transport"
            })
        );
    }

    fn session_open(
        requested_session_id: u32,
        resume_token_bytes: u32,
        session_flags: u8,
    ) -> SessionOpenMetadata {
        SessionOpenMetadata {
            requested_session_id,
            profile_id: 2,
            priority_class: crate::SessionPriorityClass::Balanced,
            session_flags,
            schema_id: 0x0000_1001,
            schema_version: 3,
            default_deadline_ms: 500,
            max_in_flight_operations: 8,
            lease_ttl_hint_ms: 30_000,
            resume_token_bytes,
            auth_bytes: 0,
            session_extension_bytes: 0,
            client_session_tag: 1,
        }
    }

    fn session_ack(
        session_id: u32,
        session_status: SessionStatus,
        session_flags_ack: u32,
        resume_window_ms: u32,
        resume_token_bytes: u32,
        session_error_code: u32,
    ) -> SessionOpenAckMetadata {
        SessionOpenAckMetadata {
            session_id,
            accepted_profile_id: 2,
            accepted_priority_class: crate::SessionPriorityClass::Balanced,
            schema_id: 0x0000_1001,
            schema_version: 3,
            granted_operation_credit: 4,
            max_in_flight_operations: 8,
            lease_ttl_ms: 30_000,
            resume_window_ms,
            resume_token_bytes,
            session_status,
            session_extension_bytes: 0,
            server_session_tag: 7,
            route_scope_id: 0,
            session_error_code,
            session_flags_ack,
        }
    }
}