feat(client): support AllowDirect responses in cursor_prep_exec

[Tawxyn]

sp_cursorprepexec may take the AllowDirect fast path: a nonzero prepared handle, @cursor == 0, and result sets streamed inline during the RPC response. Previously that response errored ("zero @cursor") and the inline rows were discarded.

cursor_prep_exec now returns a typed CursorPrepExecOutcome distinguishing a normal PreparedCursor from owned DirectResults, preserving every direct result set's metadata and rows in order and retaining correct prepared-handle cleanup. into_cursor/into_direct are lossless (hand back the non-matching variant intact so no handle leaks).

Adds a row-preserving RPC collector (collect_rpc_result_sets), unit tests for the collector and outcome logic, an offline server-smol harness test, and a live SQL Server multi-result test.

BREAKING CHANGE: cursor_prep_exec now returns CursorPrepExecOutcome instead of PreparedCursor.

[hippiehunter]

• Direct execution is detected using the wrong signal. The PR only recognizes @cursor == 0, while SQL Server signals fallback with INFO 16954 and may omit cursor OUT parameters. The collector currently ignores INFO tokens, so the response becomes an error or a
  bogus cursor parsed from the next positional output. See collector

  tiberius/src/client/rpc_response.rs

  Lines 252 to 289 in 0cb2028

  	while let Some(token) = stream.try_next().await? {
  	match token {
  	ReceivedToken::NewResultset(meta) => {
  	flush(&mut results, &mut columns, &mut current);
  	columns = Some(Arc::new(meta.columns().collect::<Vec<_>>()));
  	}
  	ReceivedToken::Row(data) => {
  	if let Some(cols) = &columns {
  	current.push(Row {
  	columns: cols.clone(),
  	data,
  	// The index this set will occupy once flushed.
  	result_index: results.len(),
  	});
  	}
  	}
  	ReceivedToken::ReturnValue(rv) => outputs.push(rv.into()),
  	ReceivedToken::ReturnStatus(s) => status = Some(s),
  	ReceivedToken::Error(e) => {
  	if last_error.is_none() {
  	last_error = Some(crate::Error::Server(e));
  	}
  	}
  	ReceivedToken::DoneInProc(done) => {
  	// A DoneInProc without `More` closes one statement's result set
  	// inside the proc; more tokens (ReturnValue, DoneProc) follow.
  	if !done.status().contains(DoneStatus::More) {
  	flush(&mut results, &mut columns, &mut current);
  	}
  	}
  	ReceivedToken::DoneProc(done) | ReceivedToken::Done(done) => {
  	if !done.status().contains(DoneStatus::More) {
  	flush(&mut results, &mut columns, &mut current);
  	break;
  	}
  	}
  	_ => {}
  	}

  outcome parsing
  

  tiberius/src/client/cursor.rs

  Lines 921 to 954 in 0cb2028

  	// A zero @cursor (with a valid prepared handle) is the AllowDirect signal:
  	// the server executed directly and streamed the result sets inline. A
  	// nonzero or absent @cursor falls through to the normal cursor path.
  	match lookup_named("cursor").or_else(|| by_pos(1)) {
  	Some(0) => {
  	let scrollopt = lookup_named("scrollopt").or_else(|| by_pos(2)).unwrap_or(0);
  	let ccopt = lookup_named("ccopt").or_else(|| by_pos(3)).unwrap_or(0);
  	let row_count = lookup_named("rowcount").or_else(|| by_pos(4)).unwrap_or(-1);
  	let results = result_sets
  	.into_iter()
  	.map(|rs| DirectResultSet {
  	columns: rs.columns,
  	rows: rs.rows,
  	})
  	.collect();
  	Ok(CursorPrepExecOutcome::Direct(DirectResults {
  	prepared_handle: PreparedHandle::from_i32(prepared_handle),
  	results,
  	scrollopt: i32_to_scroll_flags(scrollopt),
  	ccopt: i32_to_cc_flags(ccopt),
  	row_count,
  	released: false,
  	}))
  	}
  	_ => {
  	// Preserve the pre-AllowDirect behaviour: seed the cursor's cached
  	// metadata from the first result set (only non-empty sets are
  	// buffered, so `first` is the first non-empty COLMETADATA).
  	let metadata = result_sets.first().map(|rs| (*rs.columns).clone());
  	Ok(CursorPrepExecOutcome::Cursor(prepared_cursor_from_outputs(
  	outputs, metadata,
  	)?))

  and the official Microsoft driver’s handling
  https://github.com/microsoft/mssql-jdbc/blob/main/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerStatement.java#L1974-L1988

• Direct handles are released with the wrong RPC. DirectResults::unprepare sends sp_cursorunprepare; directly executed handles require sp_unprepare. This will fail cleanup and leak the handle until disconnect. Microsoft’s driver explicitly chooses between them
  based on direct execution. See PR cleanup

  tiberius/src/client/cursor.rs

  Lines 462 to 480 in 0cb2028

  	/// Release the prepared handle via `sp_cursorunprepare`, returning the
  	/// owned result sets.
  	///
  	/// There is no cursor to close, so this only unprepares the statement.
  	pub async fn unprepare<S>(
  	mut self,
  	client: &mut Client<S>,
  	) -> crate::Result<Vec<DirectResultSet>>
  	where
  	S: AsyncRead + AsyncWrite + Unpin + Send,
  	{
  	client.connection.flush_stream().await?;
  	let handle_param = build_cursorunprepare_param(self.prepared_handle);
  	client
  	.send_rpc(RpcProcId::CursorUnprepare, vec![handle_param])
  	.await?;
  	self.released = true;
  	collect_rpc_outputs(&mut client.connection).await?;
  	Ok(std::mem::take(&mut self.results))

  and Microsoft’s cleanup logic
  https://github.com/microsoft/mssql-jdbc/blob/main/src/main/java/com/microsoft/sqlserver/jdbc/SQLServerPreparedStatement.java#L388-L395