diff --git a/.github/workflows/pr-code-security.yml b/.github/workflows/pr-code-security.yml
deleted file mode 100644
index fde10666c..000000000
--- a/.github/workflows/pr-code-security.yml
+++ /dev/null
@@ -1,16 +0,0 @@
-name: PR Code Security
-
-on:
-  pull_request:
-    branches: [main]
-
-jobs:
-  secret-detection:
-    name: Secret Detection
-    if: github.event_name == 'pull_request'
-    uses: prisma/.github/.github/workflows/secret_detection.yml@main
-    secrets: inherit
-  code-scanning:
-    name: Code Scanning
-    if: github.event_name == 'pull_request'
-    uses: prisma/.github/.github/workflows/code_scanning.yml@main
diff --git a/.github/workflows/security.yml b/.github/workflows/security.yml
new file mode 100644
index 000000000..ca9357e57
--- /dev/null
+++ b/.github/workflows/security.yml
@@ -0,0 +1,29 @@
+name: Security audit
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+  schedule:
+    # Re-run weekly so newly-published advisories are caught even without a push.
+    - cron: "0 6 * * 1"
+
+permissions:
+  contents: read
+
+jobs:
+  cargo-deny:
+    name: cargo-deny (advisories, bans, sources)
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+      - uses: dtolnay/rust-toolchain@e97e2d8cc328f1b50210efc529dca0028893a2d9 # v1
+        with:
+          toolchain: stable
+      # Run cargo-deny on the runner directly (the musl container action conflicts with the repo's rust-toolchain file).
+      - name: Install cargo-deny
+        uses: taiki-e/install-action@cargo-deny
+      - name: Check advisories, bans, sources
+        run: cargo deny check advisories bans sources
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index bef994908..390041d2c 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -87,6 +87,25 @@ jobs:
       - name: Install dependencies
         run: sudo apt install -y openssl libkrb5-dev
 
+      - name: Wait for SQL Server
+        # A listening port is not readiness: SQL Server binds 1433 before the SA
+        # login and databases finish initializing, so tests started too early race
+        # it and hit sporadic connection/login failures. Gate on an authenticated
+        # `SELECT 1` from a throwaway mssql-tools container (works uniformly across
+        # the full server images and azure-sql-edge, which ships no in-box sqlcmd).
+        run: |
+          pw='<YourStrong@Passw0rd>'
+          for _ in $(seq 1 60); do
+            if docker run --rm --network host mcr.microsoft.com/mssql-tools \
+                 /opt/mssql-tools/bin/sqlcmd -S localhost,1433 -U SA -P "$pw" -Q "SELECT 1" >/dev/null 2>&1; then
+              echo "SQL Server ready (authenticated login succeeded)"; exit 0
+            fi
+            sleep 3
+          done
+          echo "SQL Server did not accept an authenticated login in time" >&2
+          docker compose -f docker-compose.yml logs mssql-${{matrix.database}} || true
+          exit 1
+
       - name: Run tests
         run: cargo test ${{matrix.features}}
 
diff --git a/Cargo.toml b/Cargo.toml
index 1ff2ba788..b0ad8e6e6 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -166,12 +166,12 @@ path = "./runtimes-macro"
 [dev-dependencies]
 names = "0.14"
 anyhow = "1"
-env_logger = "0.9"
+env_logger = "0.11"
 azure_identity = "0.20.0"
 url = "2.2.2"
 reqwest = "0.12"
 paste = "1.0"
-indicatif = "0.17"
+indicatif = "0.18"
 chrono = "0.4.38"
 indoc = "1.0.7"
 
diff --git a/deny.toml b/deny.toml
new file mode 100644
index 000000000..d384d581a
--- /dev/null
+++ b/deny.toml
@@ -0,0 +1,31 @@
+# Run locally with:  cargo deny check advisories bans sources
+# CI runs the same via .github/workflows/security.yml.
+#
+# Policy: a vulnerability or yanked crate in the default-built graph fails the build; ignored advisories are only reachable via dev-deps or opt-in features.
+
+[advisories]
+yanked = "deny"
+ignore = [
+    # async-std is discontinued upstream. In tiberius it is reachable ONLY through
+    # the opt-in `sql-browser-async-std` feature (and dev-deps); it is not part of
+    # the default build. Tracked for migration to smol/tokio.
+    { id = "RUSTSEC-2025-0052", reason = "async-std: opt-in `sql-browser-async-std` feature + dev-deps only; not in the default shipped graph" },
+
+    # The following are ALL dev-dependency-only (test harness + the aad-auth
+    # example) and are never compiled into the published library.
+    { id = "RUSTSEC-2024-0375", reason = "atty: dev-dependency only (via `names` -> clap 3); not shipped" },
+    { id = "RUSTSEC-2024-0370", reason = "proc-macro-error: dev-dependency only (via `names` -> clap 3); not shipped" },
+    { id = "RUSTSEC-2024-0384", reason = "instant: transitive dev-dependency only; not shipped" },
+    { id = "RUSTSEC-2024-0436", reason = "paste: dev/test only (tests/bulk.rs + azure_identity example); not shipped" },
+    { id = "RUSTSEC-2026-0174", reason = "http-types: dev-only via azure_identity in the aad-auth example; not shipped" },
+    { id = "RUSTSEC-2026-0275", reason = "azure_core: dev-only via azure_identity in the aad-auth example; not shipped (the crate never handles AAD tokens itself)" },
+]
+
+[bans]
+multiple-versions = "warn"
+wildcards = "allow"
+
+[sources]
+unknown-registry = "deny"
+unknown-git = "deny"
+allow-registry = ["https://github.com/rust-lang/crates.io-index"]
diff --git a/src/client/config.rs b/src/client/config.rs
index fff68bc15..57374f8ce 100644
--- a/src/client/config.rs
+++ b/src/client/config.rs
@@ -385,7 +385,7 @@ pub(crate) trait ConfigString {
     fn readonly(&self) -> bool {
         self.dict()
             .get("applicationintent")
-            .filter(|val| *val == "ReadOnly")
+            .filter(|val| val.trim().eq_ignore_ascii_case("ReadOnly"))
             .is_some()
     }
 }
diff --git a/src/client/config/ado_net.rs b/src/client/config/ado_net.rs
index 94df9ca38..018f92da7 100644
--- a/src/client/config/ado_net.rs
+++ b/src/client/config/ado_net.rs
@@ -484,4 +484,25 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn application_intent_readonly_parsing() -> crate::Result<()> {
+        // Exact spelling from the ADO.NET connection string.
+        let ado: AdoNetConfig = "ApplicationIntent=ReadOnly".parse()?;
+        assert!(ado.readonly());
+
+        // ADO.NET treats the value case-insensitively.
+        let ado: AdoNetConfig = "applicationintent=readonly".parse()?;
+        assert!(ado.readonly());
+
+        // ReadWrite (the default) must not request read-only intent.
+        let ado: AdoNetConfig = "ApplicationIntent=ReadWrite".parse()?;
+        assert!(!ado.readonly());
+
+        // Absent altogether.
+        let ado: AdoNetConfig = "server=tcp:localhost,1433".parse()?;
+        assert!(!ado.readonly());
+
+        Ok(())
+    }
 }
diff --git a/src/client/connection.rs b/src/client/connection.rs
index 09d372561..26701d165 100644
--- a/src/client/connection.rs
+++ b/src/client/connection.rs
@@ -94,7 +94,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connection<S> {
             .prelogin(config.encryption, fed_auth_required)
             .await?;
 
-        let encryption = prelogin.negotiated_encryption(config.encryption);
+        let encryption = prelogin.negotiated_encryption(config.encryption)?;
 
         let connection = connection.tls_handshake(&config, encryption).await?;
 
@@ -285,7 +285,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connection<S> {
     /// Defines the login record rules with SQL Server. Authentication with
     /// connection options.
     #[allow(clippy::too_many_arguments)]
-    async fn login<'a>(
+    async fn login(
         mut self,
         auth: AuthMethod,
         encryption: EncryptionLevel,
@@ -445,7 +445,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connection<S> {
         encryption: EncryptionLevel,
     ) -> crate::Result<Self> {
         if encryption != EncryptionLevel::NotSupported {
-            event!(Level::INFO, "Performing a TLS handshake");
+            event!(Level::DEBUG, "Performing a TLS handshake");
 
             let Self {
                 transport, context, ..
@@ -458,7 +458,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connection<S> {
             };
 
             stream.get_mut().handshake_complete();
-            event!(Level::INFO, "TLS handshake successful");
+            event!(Level::DEBUG, "TLS handshake successful");
 
             let transport = Framed::new(MaybeTlsStream::Tls(stream), PacketCodec);
 
@@ -484,7 +484,9 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connection<S> {
         feature = "native-tls",
         feature = "vendored-openssl"
     )))]
-    async fn tls_handshake(self, _: &Config, _: EncryptionLevel) -> crate::Result<Self> {
+    async fn tls_handshake(self, config: &Config, _: EncryptionLevel) -> crate::Result<Self> {
+        check_tls_backend_available(config.encryption)?;
+
         event!(
             Level::WARN,
             "TLS encryption is not enabled. All traffic including the login credentials are not encrypted."
@@ -498,6 +500,51 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> Connection<S> {
     }
 }
 
+/// Returns an error when the user requested encryption but no TLS backend was
+/// compiled in. Without this check, a `Required`/`On` encryption request would
+/// silently fall back to an unencrypted connection.
+#[cfg(not(any(
+    feature = "rustls",
+    feature = "native-tls",
+    feature = "vendored-openssl"
+)))]
+fn check_tls_backend_available(encryption: EncryptionLevel) -> crate::Result<()> {
+    if let EncryptionLevel::On | EncryptionLevel::Required = encryption {
+        return Err(crate::Error::Tls(
+            "TLS encryption was requested but the crate was compiled without a TLS backend. \
+             Enable one of the `native-tls`, `rustls` or `vendored-openssl` features."
+                .to_string(),
+        ));
+    }
+
+    Ok(())
+}
+
+#[cfg(all(
+    test,
+    not(any(
+        feature = "rustls",
+        feature = "native-tls",
+        feature = "vendored-openssl"
+    ))
+))]
+mod tests {
+    use super::check_tls_backend_available;
+    use crate::EncryptionLevel;
+
+    #[test]
+    fn requested_encryption_without_tls_backend_errors() {
+        assert!(check_tls_backend_available(EncryptionLevel::Required).is_err());
+        assert!(check_tls_backend_available(EncryptionLevel::On).is_err());
+    }
+
+    #[test]
+    fn no_encryption_without_tls_backend_is_ok() {
+        assert!(check_tls_backend_available(EncryptionLevel::Off).is_ok());
+        assert!(check_tls_backend_available(EncryptionLevel::NotSupported).is_ok());
+    }
+}
+
 impl<S: AsyncRead + AsyncWrite + Unpin + Send> Stream for Connection<S> {
     type Item = crate::Result<Packet>;
 
diff --git a/src/client/tls.rs b/src/client/tls.rs
index 3c8ff9bd7..cad84ba35 100644
--- a/src/client/tls.rs
+++ b/src/client/tls.rs
@@ -215,7 +215,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> AsyncRead for TlsPreloginWrapper<
             }
 
             let header = PacketHeader::decode(&mut BytesMut::from(&inner.header_buf[..]))
-                .map_err(|err| io::Error::new(io::ErrorKind::Other, err))?;
+                .map_err(io::Error::other)?;
 
             // We only get pre-login packets in the handshake process.
             assert_eq!(header.r#type(), PacketType::PreLogin);
diff --git a/src/client/tls_stream/native_tls_stream.rs b/src/client/tls_stream/native_tls_stream.rs
index cf5591d80..73cd10595 100644
--- a/src/client/tls_stream/native_tls_stream.rs
+++ b/src/client/tls_stream/native_tls_stream.rs
@@ -19,12 +19,12 @@ pub(crate) async fn create_tls_stream<S: AsyncRead + AsyncWrite + Unpin + Send>(
             if let Ok(buf) = fs::read(path) {
                 let cert = match path.extension() {
                         Some(ext)
-                        if ext.to_ascii_lowercase() == "pem"
-                            || ext.to_ascii_lowercase() == "crt" =>
+                        if ext.eq_ignore_ascii_case("pem")
+                            || ext.eq_ignore_ascii_case("crt") =>
                             {
                                 Some(Certificate::from_pem(&buf)?)
                             }
-                        Some(ext) if ext.to_ascii_lowercase() == "der" => {
+                        Some(ext) if ext.eq_ignore_ascii_case("der") => {
                             Some(Certificate::from_der(&buf)?)
                         }
                         Some(_) | None => return Err(Error::Io {
@@ -52,7 +52,7 @@ pub(crate) async fn create_tls_stream<S: AsyncRead + AsyncWrite + Unpin + Send>(
             builder = builder.use_sni(false);
         }
         TrustConfig::Default => {
-            event!(Level::INFO, "Using default trust configuration.");
+            event!(Level::DEBUG, "Using default trust configuration.");
         }
     }
 
diff --git a/src/client/tls_stream/opentls_tls_stream.rs b/src/client/tls_stream/opentls_tls_stream.rs
index 1f028669e..fa8009a65 100644
--- a/src/client/tls_stream/opentls_tls_stream.rs
+++ b/src/client/tls_stream/opentls_tls_stream.rs
@@ -52,7 +52,7 @@ pub(crate) async fn create_tls_stream<S: AsyncRead + AsyncWrite + Unpin + Send>(
             builder = builder.use_sni(false);
         }
         TrustConfig::Default => {
-            event!(Level::INFO, "Using default trust configuration.");
+            event!(Level::DEBUG, "Using default trust configuration.");
         }
     }
 
diff --git a/src/client/tls_stream/rustls_tls_stream.rs b/src/client/tls_stream/rustls_tls_stream.rs
index 7cf577ad7..88871ba07 100644
--- a/src/client/tls_stream/rustls_tls_stream.rs
+++ b/src/client/tls_stream/rustls_tls_stream.rs
@@ -98,7 +98,7 @@ fn get_server_name(config: &Config) -> crate::Result<ServerName<'static>> {
 
 impl<S: AsyncRead + AsyncWrite + Unpin + Send> TlsStream<S> {
     pub(super) async fn new(config: &Config, stream: S) -> crate::Result<Self> {
-        event!(Level::INFO, "Performing a TLS handshake");
+        event!(Level::DEBUG, "Performing a TLS handshake");
 
         // Negotiate TLS 1.2 or 1.3.
         let builder = ClientConfig::builder_with_provider(Arc::new(aws_lc_rs::default_provider()))
@@ -168,7 +168,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send> TlsStream<S> {
                     .with_no_client_auth()
             }
             TrustConfig::Default => {
-                event!(Level::INFO, "Using default trust configuration.");
+                event!(Level::DEBUG, "Using default trust configuration.");
                 builder.with_native_roots().with_no_client_auth()
             }
         };
diff --git a/src/from_sql.rs b/src/from_sql.rs
index 8498fa01c..f8bedff16 100644
--- a/src/from_sql.rs
+++ b/src/from_sql.rs
@@ -60,7 +60,7 @@ where
 from_sql!(bool: ColumnData::Bit(val) => (*val, val));
 from_sql!(u8: ColumnData::U8(val) => (*val, val), ColumnData::I32(None) => (None, None));
 from_sql!(i16: ColumnData::I16(val) => (*val, val), ColumnData::U8(None) => (None, None), ColumnData::I32(None) => (None, None));
-from_sql!(i32: ColumnData::I32(val) => (*val, val), ColumnData::U8(None) => (None, None));
+from_sql!(i32: ColumnData::I32(val) => (*val, val), ColumnData::I16(val) => (val.map(i32::from), val.map(i32::from)), ColumnData::U8(None) => (None, None));
 from_sql!(i64: ColumnData::I64(val) => (*val, val), ColumnData::U8(None) => (None, None), ColumnData::I32(None) => (None, None));
 from_sql!(f32: ColumnData::F32(val) => (*val, val));
 from_sql!(f64: ColumnData::F64(val) => (*val, val));
@@ -132,3 +132,22 @@ impl<'a> FromSql<'a> for &'a [u8] {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn i16_column_converts_to_i32() {
+        let data = ColumnData::I16(Some(8));
+        assert_eq!(Some(8i32), i32::from_sql(&data).unwrap());
+        assert_eq!(Some(8i32), i32::from_sql_owned(data).unwrap());
+    }
+
+    #[test]
+    fn null_i16_column_converts_to_i32() {
+        let data = ColumnData::I16(None);
+        assert_eq!(None, i32::from_sql(&data).unwrap());
+        assert_eq!(None, i32::from_sql_owned(ColumnData::I16(None)).unwrap());
+    }
+}
diff --git a/src/lib.rs b/src/lib.rs
index 882f5ad36..1115a5e2a 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -156,11 +156,11 @@
 //! Tiberius supports different [ways of authentication] to the SQL Server:
 //!
 //! - SQL Server authentication uses the facilities of the database to
-//! authenticate the user.
+//!   authenticate the user.
 //! - On Windows, you can authenticate using the currently logged in user or
-//! specified Windows credentials.
+//!   specified Windows credentials.
 //! - If enabling the `integrated-auth-gssapi` feature, it is possible to login
-//! with the currently active Kerberos credentials.
+//!   with the currently active Kerberos credentials.
 //!
 //! ## AAD(Azure Active Directory) Authentication
 //!
diff --git a/src/macros.rs b/src/macros.rs
index 35f24228f..cbe0453e1 100644
--- a/src/macros.rs
+++ b/src/macros.rs
@@ -19,7 +19,10 @@ macro_rules! uint_enum {
             type Error = ();
             fn try_from(n: u8) -> ::std::result::Result<$ty, ()> {
                 match n {
-                    $( x if x == $ty::$variant as u8 => Ok($ty::$variant), )*
+                    // Generic macro codegen: the `as u8` cast is compared against a `u8`
+                    // input, so wider enum variants can never match here (they fall through
+                    // to `Err`). The truncation is intentional and harmless.
+                    $( #[allow(clippy::cast_enum_truncation)] x if x == $ty::$variant as u8 => Ok($ty::$variant), )*
                     _ => Err(()),
                 }
             }
diff --git a/src/query.rs b/src/query.rs
index 86e949996..790052b4d 100644
--- a/src/query.rs
+++ b/src/query.rs
@@ -69,7 +69,7 @@ impl<'a> Query<'a> {
     /// [`ToSql`]: trait.ToSql.html
     /// [`FromSql`]: trait.FromSql.html
     /// [`Client#execute`]: struct.Client.html#method.execute
-    pub async fn execute<'b, S>(self, client: &'b mut Client<S>) -> crate::Result<ExecuteResult>
+    pub async fn execute<S>(self, client: &mut Client<S>) -> crate::Result<ExecuteResult>
     where
         S: AsyncRead + AsyncWrite + Unpin + Send,
     {
diff --git a/src/row.rs b/src/row.rs
index 5441be700..873802028 100644
--- a/src/row.rs
+++ b/src/row.rs
@@ -260,14 +260,17 @@ where
 }
 
 impl QueryIdx for usize {
-    fn idx(&self, _row: &Row) -> Option<usize> {
-        Some(*self)
+    fn idx(&self, row: &Row) -> Option<usize> {
+        (*self < row.columns.len()).then_some(*self)
     }
 }
 
 impl QueryIdx for &str {
     fn idx(&self, row: &Row) -> Option<usize> {
-        row.columns.iter().position(|c| c.name() == *self)
+        // Allow matching a column selected with a Rust raw identifier (e.g.
+        // `r#type`) against the plain SQL column name (`type`).
+        let name = self.strip_prefix("r#").unwrap_or(self);
+        row.columns.iter().position(|c| c.name() == name)
     }
 }
 
@@ -423,3 +426,51 @@ impl IntoIterator for Row {
         self.data.into_iter()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn make_row() -> Row {
+        let columns = Arc::new(vec![
+            Column::new("foo".to_string(), ColumnType::Int4),
+            Column::new("type".to_string(), ColumnType::Int4),
+        ]);
+
+        let mut data = TokenRow::new();
+        data.push(ColumnData::I32(Some(1)));
+        data.push(ColumnData::I32(Some(2)));
+
+        Row {
+            columns,
+            data,
+            result_index: 0,
+        }
+    }
+
+    // Regression test for #211: an out-of-range usize index must not panic.
+    #[test]
+    fn try_get_out_of_range_index_returns_none() {
+        let row = make_row();
+
+        assert_eq!(None, 2usize.idx(&row));
+        assert_eq!(Some(0), 0usize.idx(&row));
+
+        let value: crate::Result<Option<i32>> = row.try_get(5usize);
+        assert!(value.is_err());
+    }
+
+    // Regression test for #382: a raw-identifier column name (`r#type`) must
+    // match the plain SQL column name (`type`).
+    #[test]
+    fn raw_identifier_column_name_matches() {
+        let row = make_row();
+
+        assert_eq!(Some(1), "type".idx(&row));
+        assert_eq!(Some(1), "r#type".idx(&row));
+        assert_eq!(Some(0), "r#foo".idx(&row));
+        assert_eq!(None, "r#missing".idx(&row));
+
+        assert_eq!(Some(2i32), row.get::<i32, _>("r#type"));
+    }
+}
diff --git a/src/tds/codec/column_data.rs b/src/tds/codec/column_data.rs
index fecd83f75..054d10a2e 100644
--- a/src/tds/codec/column_data.rs
+++ b/src/tds/codec/column_data.rs
@@ -275,6 +275,15 @@ impl<'a> Encode<BytesMutWithTypeInfo<'a>> for ColumnData<'a> {
                 dst.extend_from_slice(&header);
                 dst.put_f64_le(val);
             }
+            (ColumnData::F64(opt), Some(TypeInfo::VarLenSized(vlc)))
+                if vlc.r#type() == VarLenType::Money =>
+            {
+                if let Some(val) = opt {
+                    money::encode(dst, vlc.len(), val);
+                } else {
+                    dst.put_u8(0);
+                }
+            }
             (ColumnData::Guid(opt), Some(TypeInfo::VarLenSized(vlc)))
                 if vlc.r#type() == VarLenType::Guid =>
             {
@@ -424,6 +433,57 @@ impl<'a> Encode<BytesMutWithTypeInfo<'a>> for ColumnData<'a> {
                     dst.put_u64_le(0xffffffffffffffff)
                 }
             }
+            (ColumnData::String(opt), Some(TypeInfo::VarLenSized(vlc)))
+                if vlc.r#type() == VarLenType::Text || vlc.r#type() == VarLenType::NText =>
+            {
+                if let Some(str) = opt {
+                    // TEXT/NTEXT row values carry a text pointer and a timestamp
+                    // ahead of the payload. The server ignores the values we
+                    // supply on bulk-load, so send a fixed-size dummy pointer
+                    // and timestamp.
+                    dst.put_u8(16); // text pointer length
+                    dst.extend_from_slice(&[0u8; 16]); // text pointer
+                    dst.extend_from_slice(&[0u8; 8]); // timestamp
+
+                    if vlc.r#type() == VarLenType::Text {
+                        // single-byte character data, encoded with the column collation
+                        let mut encoder =
+                            vlc.collation().as_ref().unwrap().encoding()?.new_encoder();
+                        let len = encoder
+                            .max_buffer_length_from_utf8_without_replacement(str.len())
+                            .unwrap();
+                        let mut bytes = Vec::with_capacity(len);
+                        let (res, _) = encoder.encode_from_utf8_to_vec_without_replacement(
+                            str.as_ref(),
+                            &mut bytes,
+                            true,
+                        );
+                        if let encoding_rs::EncoderResult::Unmappable(_) = res {
+                            return Err(crate::Error::Encoding("unrepresentable character".into()));
+                        }
+
+                        dst.put_u32_le(bytes.len() as u32);
+                        dst.extend_from_slice(bytes.as_slice());
+                    } else {
+                        // NTEXT: UCS-2/UTF-16LE data
+                        let len_pos = dst.len();
+                        dst.put_u32_le(0u32);
+
+                        let mut length = 0u32;
+                        for chr in str.encode_utf16() {
+                            length += 2;
+                            dst.put_u16_le(chr);
+                        }
+
+                        let dst: &mut [u8] = dst.borrow_mut();
+                        let bytes = length.to_le_bytes();
+                        dst[len_pos..len_pos + 4].copy_from_slice(&bytes);
+                    }
+                } else {
+                    // NULL: zero-length text pointer
+                    dst.put_u8(0);
+                }
+            }
             (ColumnData::String(Some(ref s)), None) if s.len() <= 4000 => {
                 dst.put_u8(VarLenType::NVarchar as u8);
                 dst.put_u16_le(8000);
@@ -663,6 +723,15 @@ impl<'a> Encode<BytesMutWithTypeInfo<'a>> for ColumnData<'a> {
                 dst.put_u8(0);
                 xml.into_owned().encode(&mut *dst)?;
             }
+            (ColumnData::Numeric(opt), Some(TypeInfo::VarLenSized(vlc)))
+                if vlc.r#type() == VarLenType::Money =>
+            {
+                if let Some(num) = opt {
+                    money::encode(dst, vlc.len(), f64::from(num));
+                } else {
+                    dst.put_u8(0);
+                }
+            }
             (ColumnData::Numeric(opt), Some(TypeInfo::VarLenSizedPrecision { ty, scale, .. }))
                 if ty == &VarLenType::Numericn || ty == &VarLenType::Decimaln =>
             {
diff --git a/src/tds/codec/column_data/int.rs b/src/tds/codec/column_data/int.rs
index f9d51da48..bb9271fe5 100644
--- a/src/tds/codec/column_data/int.rs
+++ b/src/tds/codec/column_data/int.rs
@@ -1,4 +1,4 @@
-use crate::{sql_read_bytes::SqlReadBytes, ColumnData};
+use crate::{sql_read_bytes::SqlReadBytes, ColumnData, Error};
 
 pub(crate) async fn decode<R>(src: &mut R, type_len: usize) -> crate::Result<ColumnData<'static>>
 where
@@ -15,7 +15,11 @@ where
         (2, _) => ColumnData::I16(Some(src.read_i16_le().await?)),
         (4, _) => ColumnData::I32(Some(src.read_i32_le().await?)),
         (8, _) => ColumnData::I64(Some(src.read_i64_le().await?)),
-        _ => unimplemented!(),
+        _ => {
+            return Err(Error::Protocol(
+                format!("invalid integer length: {}", recv_len).into(),
+            ))
+        }
     };
 
     Ok(res)
diff --git a/src/tds/codec/column_data/money.rs b/src/tds/codec/column_data/money.rs
index 5627983d4..089ebe412 100644
--- a/src/tds/codec/column_data/money.rs
+++ b/src/tds/codec/column_data/money.rs
@@ -1,4 +1,26 @@
 use crate::{error::Error, sql_read_bytes::SqlReadBytes, ColumnData};
+use bytes::BufMut;
+
+/// Encode an `f64` as a money/smallmoney value into `dst`, prefixed with a
+/// single length byte (as expected for a nullable `Money`/`Moneyn` column in a
+/// bulk-load row). `max_len` is the column's declared length (8 for `money`,
+/// 4 for `smallmoney`). Money is stored on the wire as a scaled integer
+/// (value * 10_000).
+pub(crate) fn encode<B>(dst: &mut B, max_len: usize, val: f64)
+where
+    B: BufMut,
+{
+    if max_len == 4 {
+        dst.put_u8(4);
+        dst.put_i32_le((val * 1e4).round() as i32);
+    } else {
+        dst.put_u8(8);
+        let scaled = (val * 1e4).round() as i64;
+        // money is transmitted as two 32-bit words, high word first.
+        dst.put_i32_le((scaled >> 32) as i32);
+        dst.put_u32_le(scaled as u32);
+    }
+}
 
 pub(crate) async fn decode<R>(src: &mut R, len: u8) -> crate::Result<ColumnData<'static>>
 where
@@ -22,3 +44,43 @@ where
 
     Ok(res)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Reverses the on-wire money representation the same way `decode` does,
+    // so we can assert `encode` is the exact inverse without a live server.
+    fn decode_bytes(bytes: &[u8]) -> f64 {
+        let len = bytes[0];
+        match len {
+            4 => i32::from_le_bytes(bytes[1..5].try_into().unwrap()) as f64 / 1e4,
+            8 => {
+                let high = i32::from_le_bytes(bytes[1..5].try_into().unwrap()) as i64;
+                let low = u32::from_le_bytes(bytes[5..9].try_into().unwrap()) as f64;
+                ((high << 32) as f64 + low) / 1e4
+            }
+            _ => panic!("invalid length"),
+        }
+    }
+
+    #[test]
+    fn encode_smallmoney_roundtrips() {
+        let mut buf = Vec::new();
+        encode(&mut buf, 4, 1234.5678);
+        assert_eq!(buf[0], 4);
+        assert_eq!(buf.len(), 5);
+        assert_eq!(decode_bytes(&buf), 1234.5678);
+    }
+
+    #[test]
+    fn encode_money_roundtrips() {
+        for val in [0.0, 1.0, -1.0, 1234.5678, -9999.9999, 92233720368.5477] {
+            let mut buf = Vec::new();
+            encode(&mut buf, 8, val);
+            assert_eq!(buf[0], 8);
+            assert_eq!(buf.len(), 9);
+            assert!((decode_bytes(&buf) - val).abs() < 1e-3, "val={}", val);
+        }
+    }
+}
diff --git a/src/tds/codec/column_data/var_len.rs b/src/tds/codec/column_data/var_len.rs
index 20f6a953a..06d8eeb59 100644
--- a/src/tds/codec/column_data/var_len.rs
+++ b/src/tds/codec/column_data/var_len.rs
@@ -1,4 +1,6 @@
-use crate::{sql_read_bytes::SqlReadBytes, tds::codec::VarLenContext, ColumnData, VarLenType};
+use crate::{
+    sql_read_bytes::SqlReadBytes, tds::codec::VarLenContext, ColumnData, Error, VarLenType,
+};
 
 pub(crate) async fn decode<R>(
     src: &mut R,
@@ -41,7 +43,11 @@ where
         Text => super::text::decode(src, collation).await?,
         NText => super::text::decode(src, None).await?,
         Image => super::image::decode(src).await?,
-        t => unimplemented!("{:?}", t),
+        t => {
+            return Err(Error::Protocol(
+                format!("unsupported column type: {:?}", t).into(),
+            ))
+        }
     };
 
     Ok(res)
diff --git a/src/tds/codec/decode.rs b/src/tds/codec/decode.rs
index d19fec0c9..b97766833 100644
--- a/src/tds/codec/decode.rs
+++ b/src/tds/codec/decode.rs
@@ -53,10 +53,7 @@ impl Decoder for PacketCodec {
                 if buf.is_empty() {
                     Ok(None)
                 } else {
-                    Err(
-                        std::io::Error::new(std::io::ErrorKind::Other, "bytes remaining on stream")
-                            .into(),
-                    )
+                    Err(std::io::Error::other("bytes remaining on stream").into())
                 }
             }
         }
diff --git a/src/tds/codec/header.rs b/src/tds/codec/header.rs
index fcee5b09f..cfabf2c55 100644
--- a/src/tds/codec/header.rs
+++ b/src/tds/codec/header.rs
@@ -57,7 +57,7 @@ pub(crate) struct PacketHeader {
 
 impl PacketHeader {
     pub fn new(length: usize, id: u8) -> PacketHeader {
-        assert!(length <= u16::max_value() as usize);
+        assert!(length <= u16::MAX as usize);
         PacketHeader {
             ty: PacketType::TDSv7Login,
             status: PacketStatus::ResetConnection,
diff --git a/src/tds/codec/login.rs b/src/tds/codec/login.rs
index 265db381e..bab14d736 100644
--- a/src/tds/codec/login.rs
+++ b/src/tds/codec/login.rs
@@ -8,7 +8,7 @@ use std::{borrow::Cow, io};
 
 uint_enum! {
     #[repr(u32)]
-    #[derive(PartialOrd)]
+    #[derive(PartialOrd, Default)]
     pub enum FeatureLevel {
         SqlServerV7 = 0x70000000,
         SqlServer2000 = 0x71000000,
@@ -17,16 +17,11 @@ uint_enum! {
         SqlServer2008 = 0x730A0003,
         SqlServer2008R2 = 0x730B0003,
         /// 2012, 2014, 2016
+        #[default]
         SqlServerN = 0x74000004,
     }
 }
 
-impl Default for FeatureLevel {
-    fn default() -> Self {
-        Self::SqlServerN
-    }
-}
-
 impl FeatureLevel {
     pub fn done_row_count_bytes(self) -> u8 {
         if self as u32 >= FeatureLevel::SqlServer2005 as u32 {
@@ -558,6 +553,44 @@ mod tests {
         }
     }
 
+    #[test]
+    fn readonly_intent_sets_type_flag_bit() {
+        // The TypeFlags byte is the third of the four flag bytes, which follow
+        // the length + five u32 header fields:
+        //   4 (length) + 5 * 4 (header) = 24, then OptionFlags1, OptionFlags2,
+        //   TypeFlags at byte offset 26.
+        const TYPE_FLAGS_OFFSET: usize = 26;
+
+        let mut payload = BytesMut::new();
+        let mut login = LoginMessage::new();
+        login.readonly(true);
+        login
+            .clone()
+            .encode(&mut payload)
+            .expect("encode should succeed");
+
+        assert_eq!(
+            payload[TYPE_FLAGS_OFFSET] & LoginTypeFlag::ReadOnlyIntent as u8,
+            LoginTypeFlag::ReadOnlyIntent as u8,
+            "fReadOnlyIntent bit must be set in the encoded LOGIN7 TypeFlags byte"
+        );
+
+        // Round-trips back into the decoded message.
+        let decoded = LoginMessage::decode(&mut payload).expect("decode should succeed");
+        assert!(decoded.type_flags.contains(LoginTypeFlag::ReadOnlyIntent));
+
+        // And when not requested, the bit stays clear.
+        let mut payload = BytesMut::new();
+        let mut login = LoginMessage::new();
+        login.readonly(false);
+        login.encode(&mut payload).expect("encode should succeed");
+        assert_eq!(
+            payload[TYPE_FLAGS_OFFSET] & LoginTypeFlag::ReadOnlyIntent as u8,
+            0,
+            "fReadOnlyIntent bit must be clear when read-only intent is not requested"
+        );
+    }
+
     #[test]
     fn login_message_round_trip() {
         let mut payload = BytesMut::new();
diff --git a/src/tds/codec/pre_login.rs b/src/tds/codec/pre_login.rs
index eb4c27e60..a21f0bce6 100644
--- a/src/tds/codec/pre_login.rs
+++ b/src/tds/codec/pre_login.rs
@@ -62,18 +62,22 @@ impl PreloginMessage {
         feature = "native-tls",
         feature = "vendored-openssl"
     ))]
-    pub fn negotiated_encryption(&self, expected: EncryptionLevel) -> EncryptionLevel {
-        match (expected, self.encryption) {
+    pub fn negotiated_encryption(&self, expected: EncryptionLevel) -> Result<EncryptionLevel> {
+        let level = match (expected, self.encryption) {
             (EncryptionLevel::NotSupported, EncryptionLevel::NotSupported) => {
                 EncryptionLevel::NotSupported
             }
             (EncryptionLevel::Off, EncryptionLevel::Off) => EncryptionLevel::Off,
             (EncryptionLevel::On, EncryptionLevel::Off)
             | (EncryptionLevel::On, EncryptionLevel::NotSupported) => {
-                panic!("Server does not allow the requested encryption level.")
+                return Err(Error::Protocol(
+                    "Server does not allow the requested encryption level.".into(),
+                ))
             }
             (_, _) => EncryptionLevel::On,
-        }
+        };
+
+        Ok(level)
     }
 
     #[cfg(not(any(
@@ -81,8 +85,8 @@ impl PreloginMessage {
         feature = "native-tls",
         feature = "vendored-openssl"
     )))]
-    pub fn negotiated_encryption(&self, _: EncryptionLevel) -> EncryptionLevel {
-        EncryptionLevel::NotSupported
+    pub fn negotiated_encryption(&self, _: EncryptionLevel) -> Result<EncryptionLevel> {
+        Ok(EncryptionLevel::NotSupported)
     }
 }
 
@@ -205,7 +209,9 @@ impl Decode<BytesMut> for PreloginMessage {
                     } else if length == 4 {
                         cursor.read_u32::<BigEndian>()?
                     } else {
-                        panic!("should never happen")
+                        return Err(Error::Protocol(
+                            format!("prelogin: invalid threadid length: {}", length).into(),
+                        ));
                     }
                 }
                 // mars
@@ -240,7 +246,11 @@ impl Decode<BytesMut> for PreloginMessage {
 
                     ret.nonce = Some(data);
                 }
-                _ => panic!("unsupported prelogin token: {}", token),
+                _ => {
+                    return Err(Error::Protocol(
+                        format!("unsupported prelogin token: {}", token).into(),
+                    ))
+                }
             }
 
             cursor.set_position(old_pos);
@@ -282,4 +292,33 @@ mod tests {
 
         assert_eq!(prelogin, decoded);
     }
+
+    // #425: a server declining the requested encryption level must yield a
+    // catchable protocol error instead of panicking.
+    #[cfg(any(
+        feature = "rustls",
+        feature = "native-tls",
+        feature = "vendored-openssl"
+    ))]
+    #[test]
+    fn negotiated_encryption_rejects_declined_level() {
+        let mut prelogin = PreloginMessage::new();
+        // Server responds with an encryption level the client did not offer /
+        // that is weaker than the required `On`.
+        prelogin.encryption = EncryptionLevel::Off;
+
+        let result = prelogin.negotiated_encryption(EncryptionLevel::On);
+
+        match result {
+            Err(Error::Protocol(_)) => {}
+            other => panic!("expected Err(Error::Protocol), got {:?}", other),
+        }
+
+        // A matching, valid negotiation still succeeds.
+        prelogin.encryption = EncryptionLevel::On;
+        assert_eq!(
+            prelogin.negotiated_encryption(EncryptionLevel::On).unwrap(),
+            EncryptionLevel::On
+        );
+    }
 }
diff --git a/src/tds/codec/token/token_env_change.rs b/src/tds/codec/token/token_env_change.rs
index ecbb9612f..96d52d5a4 100644
--- a/src/tds/codec/token/token_env_change.rs
+++ b/src/tds/codec/token/token_env_change.rs
@@ -84,10 +84,10 @@ pub enum TokenEnvChange {
 impl fmt::Display for TokenEnvChange {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            Self::Database(ref old, ref new) => {
+            Self::Database(ref new, ref old) => {
                 write!(f, "Database change from '{}' to '{}'", old, new)
             }
-            Self::PacketSize(old, new) => {
+            Self::PacketSize(new, old) => {
                 write!(f, "Packet size change from '{}' to '{}'", old, new)
             }
             Self::SqlCollation { old, new } => match (old, new) {
@@ -260,3 +260,28 @@ impl TokenEnvChange {
         Ok(token)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::TokenEnvChange;
+
+    #[test]
+    fn database_display_uses_old_then_new() {
+        // Fields are stored (new, old); Display must print "from old to new".
+        let change = TokenEnvChange::Database("newdb".to_string(), "olddb".to_string());
+        assert_eq!(
+            format!("{}", change),
+            "Database change from 'olddb' to 'newdb'"
+        );
+    }
+
+    #[test]
+    fn packet_size_display_uses_old_then_new() {
+        // Fields are stored (new, old); Display must print "from old to new".
+        let change = TokenEnvChange::PacketSize(8192, 4096);
+        assert_eq!(
+            format!("{}", change),
+            "Packet size change from '4096' to '8192'"
+        );
+    }
+}
diff --git a/src/tds/codec/token/token_feature_ext_ack.rs b/src/tds/codec/token/token_feature_ext_ack.rs
index 1ba108f99..74cb3564b 100644
--- a/src/tds/codec/token/token_feature_ext_ack.rs
+++ b/src/tds/codec/token/token_feature_ext_ack.rs
@@ -1,4 +1,4 @@
-use crate::{SqlReadBytes, FEA_EXT_FEDAUTH, FEA_EXT_TERMINATOR};
+use crate::{Error, SqlReadBytes, FEA_EXT_FEDAUTH, FEA_EXT_TERMINATOR};
 use futures_util::AsyncReadExt;
 
 #[derive(Debug)]
@@ -40,12 +40,20 @@ impl TokenFeatureExtAck {
                 } else if data_len == 0 {
                     None
                 } else {
-                    panic!("invalid Feature_Ext_Ack token");
+                    return Err(Error::Protocol(
+                        format!(
+                            "invalid Feature_Ext_Ack token: invalid data length {}",
+                            data_len
+                        )
+                        .into(),
+                    ));
                 };
 
                 features.push(FeatureAck::FedAuth(FedAuthAck::SecurityToken { nonce }))
             } else {
-                unimplemented!("unsupported feature {}", feature_id)
+                return Err(Error::Protocol(
+                    format!("unsupported feature {}", feature_id).into(),
+                ));
             }
         }
 
diff --git a/src/tds/codec/token/token_row.rs b/src/tds/codec/token/token_row.rs
index b1ff16b6c..d83692c08 100644
--- a/src/tds/codec/token/token_row.rs
+++ b/src/tds/codec/token/token_row.rs
@@ -177,7 +177,7 @@ impl RowBitmap {
     where
         R: SqlReadBytes + Unpin,
     {
-        let size = (columns + 8 - 1) / 8;
+        let size = columns.div_ceil(8);
         let mut data = vec![0; size];
         src.read_exact(&mut data[0..size]).await?;
 
diff --git a/src/tds/codec/type_info.rs b/src/tds/codec/type_info.rs
index 20647d70a..4e67a179b 100644
--- a/src/tds/codec/type_info.rs
+++ b/src/tds/codec/type_info.rs
@@ -2,7 +2,7 @@ use asynchronous_codec::BytesMut;
 use bytes::BufMut;
 
 use crate::{tds::Collation, xml::XmlSchema, Error, SqlReadBytes};
-use std::{convert::TryFrom, sync::Arc, usize};
+use std::{convert::TryFrom, sync::Arc};
 
 use super::Encode;
 
diff --git a/src/tds/collation.rs b/src/tds/collation.rs
index 20367728a..ec6a5f4bb 100644
--- a/src/tds/collation.rs
+++ b/src/tds/collation.rs
@@ -48,7 +48,7 @@ impl Collation {
         res.ok_or_else(|| {
             Error::Encoding(
                 format!(
-                    "encoding: unspported encoding (LCID: {:#02x}, sort ID: {})",
+                    "encoding: unspported encoding (LCID: {:#04x}, sort ID: {})",
                     self.lcid(),
                     self.sort_id(),
                 )
@@ -74,7 +74,7 @@ impl fmt::Display for Collation {
 /// 1. (regex)replace: (.*?)\((.*?),(.*?)\) with $2 => $3
 /// 2. replace: Encoding.CP(.*?) with encoding::all::WINDOWS_$1
 /// 3. replace: Encoding.UNICODE with encoding::all::UTF16_LE
-//
+///
 /// the unimplemented!() one's are not supported by rust-encoding
 pub fn lcid_to_encoding(locale: u16) -> Option<&'static Encoding> {
     match locale {
diff --git a/src/tds/numeric.rs b/src/tds/numeric.rs
index 4f856bebb..e4eff9ceb 100644
--- a/src/tds/numeric.rs
+++ b/src/tds/numeric.rs
@@ -112,7 +112,7 @@ impl Numeric {
             #[cfg(target_endian = "big")]
             let (low_part, high_part) = (high_part, low_part);
 
-            let high_part = high_part * (u64::max_value() as u128 + 1);
+            let high_part = high_part * (u64::MAX as u128 + 1);
             low_part + high_part
         }
 
diff --git a/src/tds/stream/token.rs b/src/tds/stream/token.rs
index 35ce0658b..e8c9bc233 100644
--- a/src/tds/stream/token.rs
+++ b/src/tds/stream/token.rs
@@ -184,27 +184,27 @@ where
             _ => (),
         }
 
-        event!(Level::INFO, "{}", change);
+        event!(Level::DEBUG, "{}", change);
 
         Ok(ReceivedToken::EnvChange(change))
     }
 
     async fn get_info(&mut self) -> crate::Result<ReceivedToken> {
         let info = TokenInfo::decode(self.conn).await?;
-        event!(Level::INFO, "{}", info.message);
+        event!(Level::DEBUG, "{}", info.message);
         Ok(ReceivedToken::Info(info))
     }
 
     async fn get_login_ack(&mut self) -> crate::Result<ReceivedToken> {
         let ack = TokenLoginAck::decode(self.conn).await?;
-        event!(Level::INFO, "{} version {}", ack.prog_name, ack.version);
+        event!(Level::DEBUG, "{} version {}", ack.prog_name, ack.version);
         Ok(ReceivedToken::LoginAck(ack))
     }
 
     async fn get_feature_ext_ack(&mut self) -> crate::Result<ReceivedToken> {
         let ack = TokenFeatureExtAck::decode(self.conn).await?;
         event!(
-            Level::INFO,
+            Level::DEBUG,
             "FeatureExtAck with {} features",
             ack.features.len()
         );
@@ -247,7 +247,11 @@ where
                 TokenType::LoginAck => this.get_login_ack().await?,
                 TokenType::Sspi => this.get_sspi().await?,
                 TokenType::FeatureExtAck => this.get_feature_ext_ack().await?,
-                _ => panic!("Token {:?} unimplemented!", ty),
+                _ => {
+                    return Err(Error::Protocol(
+                        format!("Token {:?} unimplemented!", ty).into(),
+                    ))
+                }
             };
 
             Ok(Some((token, this)))
diff --git a/src/tds/time.rs b/src/tds/time.rs
index 05a1c053c..120acc4fa 100644
--- a/src/tds/time.rs
+++ b/src/tds/time.rs
@@ -27,6 +27,8 @@ pub mod chrono;
 
 #[cfg(feature = "time")]
 #[cfg_attr(feature = "docs", doc(cfg(feature = "time")))]
+// Submodule intentionally shares the name of the `time` feature/crate it wraps.
+#[allow(clippy::module_inception)]
 pub mod time;
 
 use crate::{tds::codec::Encode, SqlReadBytes};
diff --git a/src/tds/time/time.rs b/src/tds/time/time.rs
index 5a2b1cfaa..f036744bc 100644
--- a/src/tds/time/time.rs
+++ b/src/tds/time/time.rs
@@ -10,9 +10,12 @@ pub use time::{Date, Month, OffsetDateTime, PrimitiveDateTime, Time, UtcOffset};
 use crate::tds::codec::ColumnData;
 
 #[inline]
-fn from_days(days: u64, start_year: i32) -> Date {
-    Date::from_calendar_date(start_year, Month::January, 1).unwrap()
-        + Duration::from_secs(60 * 60 * 24 * days)
+fn from_days(days: i64, start_year: i32) -> Date {
+    // Use the signed `time::Duration` so that negative day offsets (dates
+    // before `start_year`, e.g. `datetime` values prior to 1900) do not
+    // overflow. Casting a negative day count into an unsigned type and
+    // multiplying it out panics with "multiply with overflow".
+    Date::from_calendar_date(start_year, Month::January, 1).unwrap() + time::Duration::days(days)
 }
 
 #[inline]
@@ -46,15 +49,15 @@ fn to_sec_fragments(from: Time) -> i64 {
 from_sql!(
     PrimitiveDateTime:
         ColumnData::SmallDateTime(ref dt) => dt.map(|dt| PrimitiveDateTime::new(
-            from_days(dt.days as u64, 1900),
+            from_days(dt.days as i64, 1900),
             from_secs(dt.seconds_fragments as u64 * 60),
         )),
         ColumnData::DateTime2(ref dt) => dt.map(|dt| PrimitiveDateTime::new(
-            from_days(dt.date.days() as u64, 1),
+            from_days(dt.date.days() as i64, 1),
             Time::from_hms(0,0,0).unwrap() + Duration::from_nanos(dt.time.increments * 10u64.pow(9 - dt.time.scale as u32))
         )),
         ColumnData::DateTime(ref dt) => dt.map(|dt| PrimitiveDateTime::new(
-            from_days(dt.days as u64, 1900),
+            from_days(dt.days as i64, 1900),
             from_sec_fragments(dt.seconds_fragments as u64)
         ));
     Time:
@@ -63,10 +66,10 @@ from_sql!(
             Time::from_hms(0,0,0).unwrap() + Duration::from_nanos(ns)
         });
     Date:
-        ColumnData::Date(ref date) => date.map(|date| from_days(date.days() as u64, 1));
+        ColumnData::Date(ref date) => date.map(|date| from_days(date.days() as i64, 1));
     OffsetDateTime:
         ColumnData::DateTimeOffset(ref dto) => dto.map(|dto| {
-            let date = from_days(dto.datetime2.date.days() as u64, 1);
+            let date = from_days(dto.datetime2.date.days() as i64, 1);
             let dt = dto.datetime2;
 
             let time = Time::from_hms(0,0,0).unwrap()
@@ -129,6 +132,50 @@ to_sql!(self_,
 from_sql!(
     PrimitiveDateTime:
     ColumnData::DateTime(ref dt) => dt.map(|dt| {
-        from_days(dt.days as u64, 1900).with_time(from_sec_fragments(dt.seconds_fragments as u64))
+        from_days(dt.days as i64, 1900).with_time(from_sec_fragments(dt.seconds_fragments as u64))
     })
 );
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // Regression test for #316: a `datetime` value with a date before 1900 has
+    // a negative day offset from the 1900 base date. This must round-trip
+    // without a "multiply with overflow" panic.
+    #[test]
+    fn from_days_handles_negative_offsets() {
+        // 1899-12-31 is one day before the 1900 base date.
+        assert_eq!(
+            from_days(-1, 1900),
+            Date::from_calendar_date(1899, Month::December, 31).unwrap()
+        );
+
+        // A date well before 1900, at the lower edge of the `datetime` range.
+        let expected = Date::from_calendar_date(1850, Month::January, 1).unwrap();
+        let days = to_days(expected, 1900);
+        assert!(
+            days < 0,
+            "expected a negative day offset for pre-1900 dates"
+        );
+
+        // Rebuilding from the (negative) day offset must not overflow.
+        assert_eq!(from_days(days, 1900), expected);
+    }
+
+    // Exercise the full decode path (`DateTime` -> `PrimitiveDateTime`) for a
+    // pre-1900 value, matching what happens when reading a `datetime` column.
+    #[test]
+    fn datetime_before_1900_decodes() {
+        let expected_date = Date::from_calendar_date(1850, Month::January, 1).unwrap();
+        let days = to_days(expected_date, 1900) as i32;
+
+        // Reconstruct the way the `from_sql!` mapping does for `ColumnData::DateTime`.
+        let dt = crate::tds::time::DateTime::new(days, 0);
+        let decoded = from_days(dt.days() as i64, 1900)
+            .with_time(from_sec_fragments(dt.seconds_fragments() as u64));
+
+        assert_eq!(decoded.date(), expected_date);
+        assert_eq!(decoded.time(), Time::from_hms(0, 0, 0).unwrap());
+    }
+}
diff --git a/tests/query.rs b/tests/query.rs
index 4cf3c62bd..0a7b120e4 100644
--- a/tests/query.rs
+++ b/tests/query.rs
@@ -40,6 +40,9 @@ async fn random_table() -> String {
 
 static DOT_CONN_STR: Lazy<String> = Lazy::new(|| CONN_STR.replace("localhost", "."));
 
+static APP_NAME_CONN_STR: Lazy<String> =
+    Lazy::new(|| format!("{};Application Name=meow", *CONN_STR));
+
 static ENCRYPTED_CONN_STR: Lazy<String> = Lazy::new(|| format!("{};encrypt=true", *CONN_STR));
 
 static PLAIN_TEXT_CONN_STR: Lazy<String> =
@@ -2685,94 +2688,62 @@ where
     Ok(())
 }
 
-#[test]
-#[cfg(feature = "sql-browser-async-std")]
-fn cyrillic_collations_should_work() -> Result<()> {
-    LOGGER_SETUP.call_once(|| {
-        env_logger::init();
-    });
-
-    async_std::task::block_on(async {
-        let mut admin = {
-            let config = tiberius::Config::from_ado_string(&CONN_STR)?;
-
-            let tcp = async_std::net::TcpStream::connect(config.get_addr()).await?;
-            tcp.set_nodelay(true)?;
-
-            tiberius::Client::connect(config, tcp).await?
-        };
+#[test_on_runtimes]
+async fn cyrillic_collations_should_work<S>(mut conn: tiberius::Client<S>) -> Result<()>
+where
+    S: AsyncRead + AsyncWrite + Unpin + Send,
+{
+    conn.simple_query(
+        "CREATE TABLE #cyrillic_test (
+            single CHAR(1)       COLLATE Cyrillic_General_CI_AS,
+            multi  VARCHAR(255)  COLLATE Cyrillic_General_CI_AS,
+            huge   TEXT          COLLATE Cyrillic_General_CI_AS
+        )",
+    )
+    .await?;
 
-        admin
-            .simple_query("CREATE DATABASE ru_test COLLATE Cyrillic_General_CI_AS")
-            .await?;
+    conn.execute(
+        "INSERT INTO #cyrillic_test (single, multi, huge) VALUES (@P1, @P2, @P3)",
+        &[
+            &"Ж",
+            &"В Советском Союзе попытки борьбы с пьянством предпринимались не единожды. Первая антиалкогольная",
+            &"Первая антиалкогольная",
+        ],
+    )
+    .await?;
 
-        {
-            let mut client = {
-                let mut config = tiberius::Config::from_ado_string(&CONN_STR)?;
-                config.database("ru_test");
-
-                let tcp = async_std::net::TcpStream::connect(config.get_addr()).await?;
-                tcp.set_nodelay(true)?;
-
-                tiberius::Client::connect(config, tcp).await?
-            };
-
-            client
-                .simple_query(
-                    "CREATE TABLE test (id INT IDENTITY PRIMARY KEY, single CHAR(1), multi VARCHAR(255), huge TEXT)",
-                )
-                .await?;
-
-            client.execute(
-                "INSERT INTO test (single, multi, huge) VALUES (@P1, @P2, @P3)",
-                &[&"Ж", &"В Советском Союзе попытки борьбы с пьянством предпринимались не единожды. Первая антиалкогольная", &"Первая антиалкогольная"]
-            ).await?;
-
-            let row = client
-                .query("SELECT single, multi, huge FROM test", &[])
-                .await?
-                .into_row()
-                .await?
-                .unwrap();
-
-            assert_eq!(Some("Ж"), row.get(0));
-            assert_eq!(Some("В Советском Союзе попытки борьбы с пьянством предпринимались не единожды. Первая антиалкогольная"), row.get(1));
-            assert_eq!(Some("Первая антиалкогольная"), row.get(2));
-        }
+    let row = conn
+        .query("SELECT single, multi, huge FROM #cyrillic_test", &[])
+        .await?
+        .into_row()
+        .await?
+        .unwrap();
 
-        admin.simple_query("DROP DATABASE ru_test").await?;
+    assert_eq!(Some("Ж"), row.get(0));
+    assert_eq!(
+        Some("В Советском Союзе попытки борьбы с пьянством предпринимались не единожды. Первая антиалкогольная"),
+        row.get(1)
+    );
+    assert_eq!(Some("Первая антиалкогольная"), row.get(2));
 
-        Ok(())
-    })
+    Ok(())
 }
 
-#[test]
-#[cfg(feature = "sql-browser-async-std")]
-fn application_name_should_be_set_correctly() -> Result<()> {
-    LOGGER_SETUP.call_once(|| {
-        env_logger::init();
-    });
-
-    async_std::task::block_on(async {
-        let mut config = tiberius::Config::from_ado_string(&CONN_STR)?;
-        config.application_name("meow");
-
-        let tcp = async_std::net::TcpStream::connect(config.get_addr()).await?;
-        tcp.set_nodelay(true)?;
-
-        let mut client = tiberius::Client::connect(config, tcp).await?;
-
-        let row = client
-            .query("SELECT APP_NAME()", &[])
-            .await?
-            .into_row()
-            .await?
-            .unwrap();
+#[test_on_runtimes(connection_string = "APP_NAME_CONN_STR")]
+async fn application_name_should_be_set_correctly<S>(mut conn: tiberius::Client<S>) -> Result<()>
+where
+    S: AsyncRead + AsyncWrite + Unpin + Send,
+{
+    let row = conn
+        .query("SELECT APP_NAME()", &[])
+        .await?
+        .into_row()
+        .await?
+        .unwrap();
 
-        assert_eq!(Some("meow"), row.get(0));
+    assert_eq!(Some("meow"), row.get(0));
 
-        Ok(())
-    })
+    Ok(())
 }
 
 #[test_on_runtimes]