netget/server/fix/
mod.rs

1//! FIX acceptor. Rust owns framing, the Logon handshake rules, the session layer and its timers;
2//! the handler admits each counterparty and answers each application message.
3pub mod actions;
4pub mod codec;
5pub mod dict;
6pub mod session;
7
8use crate::llm::action_helper::call_llm;
9use crate::llm::actions::protocol_trait::ActionResult;
10use crate::logging::emit::Log;
11use crate::protocol::{Event, SpawnContext};
12use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
13use crate::server::connection::ConnectionId;
14use crate::state::client_handles::{ClientCommand, ClientSendOutcome};
15use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
16use crate::utils::clock::Instant;
17use anyhow::Result;
18use codec::{Frame, Message};
19use serde_json::{json, Value};
20use session::Session;
21use std::net::SocketAddr;
22use std::sync::Arc;
23use std::time::Duration;
24use tokio::io::{AsyncReadExt, AsyncWriteExt};
25use tokio::net::TcpStream;
26use tokio::sync::mpsc;
27
28pub const DEFAULT_COMP_ID: &str = "NETGET";
29pub const DEFAULT_BEGIN_STRING: &str = "FIX.4.4";
30pub const LOGON_TIMEOUT: Duration = Duration::from_secs(10);
31pub const BEGIN_STRINGS: &[&str] = &["FIX.4.0", "FIX.4.1", "FIX.4.2", "FIX.4.3", "FIX.4.4"];
32/// How long to wait for the counterparty's Logout after sending ours.
33pub const LOGOUT_WAIT: Duration = Duration::from_secs(5);
34/// HeartBtInt 0 means no heartbeats; the timers then run this rarely.
35const NO_HEARTBEAT: Duration = Duration::from_secs(86_400);
36
37struct Shared {
38    ctx: SpawnContext,
39    comp_id: String,
40    begin: String,
41    logon_timeout: Duration,
42}
43
44pub fn comp_id_ok(s: &str) -> bool {
45    !s.is_empty() && s.len() <= 64 && s.bytes().all(|b| b.is_ascii_graphic())
46}
47
48pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
49    let p = ctx.startup_params.as_ref();
50    let s = |k: &str, d: &str| -> Result<String> {
51        Ok(p.map(|p| p.get_optional_string(k))
52            .transpose()?
53            .flatten()
54            .unwrap_or_else(|| d.to_owned()))
55    };
56    let comp_id = s("sender_comp_id", DEFAULT_COMP_ID)?;
57    anyhow::ensure!(
58        comp_id_ok(&comp_id),
59        "sender_comp_id is 1 to 64 printable characters without spaces"
60    );
61    let begin = s("begin_string", DEFAULT_BEGIN_STRING)?;
62    anyhow::ensure!(
63        BEGIN_STRINGS.contains(&begin.as_str()),
64        "begin_string is one of {BEGIN_STRINGS:?}"
65    );
66    let logon_timeout = p
67        .map(|p| p.get_optional_u64("logon_timeout_secs"))
68        .transpose()?
69        .flatten()
70        .unwrap_or(LOGON_TIMEOUT.as_secs());
71    anyhow::ensure!(
72        (1..=120).contains(&logon_timeout),
73        "logon_timeout_secs must be 1..=120"
74    );
75    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr()).await?;
76    let local = listener.local_addr()?;
77    Log::new(Some(&ctx.status_tx)).info(format!("FIX acceptor {comp_id} ({begin}) on {local}"));
78    let shared = Arc::new(Shared {
79        ctx: ctx.clone(),
80        comp_id,
81        begin,
82        logon_timeout: Duration::from_secs(logon_timeout),
83    });
84    let server_id = ctx.server_id;
85    let accept = tokio::spawn(async move {
86        let limiter = ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS);
87        loop {
88            let (stream, peer, permit) =
89                match accept_bounded(&listener, &limiter, b"", "FIX", Some(&shared.ctx.status_tx))
90                    .await
91                {
92                    Ok(v) => v,
93                    Err(_) => break,
94                };
95            let id = ConnectionId::new(shared.ctx.state.get_next_unified_id().await);
96            let now = Instant::now();
97            shared
98                .ctx
99                .state
100                .add_connection_to_server(
101                    server_id,
102                    ConnectionState {
103                        id,
104                        remote_addr: peer,
105                        local_addr: local,
106                        bytes_sent: 0,
107                        bytes_received: 0,
108                        packets_sent: 0,
109                        packets_received: 0,
110                        last_activity: now,
111                        status: ConnectionStatus::Active,
112                        status_changed_at: now,
113                        protocol_info: ProtocolConnectionInfo::empty(),
114                    },
115                )
116                .await;
117            let child = shared.clone();
118            shared
119                .ctx
120                .state
121                .spawn_server_task(server_id, async move {
122                    let _permit = permit;
123                    if let Err(e) = connection(&child, id, stream).await {
124                        Log::new(Some(&child.ctx.status_tx))
125                            .debug(format!("FIX connection {id}: {e:#}"));
126                    }
127                    child
128                        .ctx
129                        .state
130                        .remove_peer_handle(server_id, id.as_u32())
131                        .await;
132                    child
133                        .ctx
134                        .state
135                        .update_connection_status(server_id, id, ConnectionStatus::Closed)
136                        .await;
137                    let _ = child.ctx.status_tx.send("__UPDATE_UI__".into());
138                })
139                .await;
140        }
141    });
142    ctx.state.register_server_task(server_id, accept).await;
143    Ok(local)
144}
145
146fn outcome(ctx: &SpawnContext, id: ConnectionId, operation: &str, decision: &str) {
147    let summary = format!("FIX connection {id} operation={operation} decision={decision}");
148    let log = Log::new(Some(&ctx.status_tx));
149    if decision.starts_with("fail_closed_") || decision == "model_silent" {
150        log.error(summary);
151    } else {
152        log.info(summary);
153    }
154}
155
156/// Ask the handler; every answer it gave, or the fail-closed reason.
157async fn ask(
158    shared: &Shared,
159    id: ConnectionId,
160    event: Event,
161    operation: &str,
162) -> Result<Vec<Value>, anyhow::Error> {
163    let ctx = &shared.ctx;
164    let result = match call_llm(
165        &ctx.llm_client,
166        &ctx.state,
167        ctx.server_id,
168        Some(id),
169        &event,
170        &actions::FixProtocol,
171    )
172    .await
173    {
174        Ok(r) => r,
175        Err(e) => {
176            outcome(ctx, id, operation, "fail_closed_llm_error");
177            return Err(e);
178        }
179    };
180    if !result.failures.is_empty() {
181        outcome(ctx, id, operation, "fail_closed_invalid_reply");
182        anyhow::bail!("the handler's answer was invalid");
183    }
184    let mut answers = Vec::new();
185    let mut pending: Vec<ActionResult> = result.protocol_results.into_iter().rev().collect();
186    while let Some(r) = pending.pop() {
187        match r {
188            ActionResult::Custom { data, .. } => answers.push(data),
189            ActionResult::Multiple(items) => pending.extend(items.into_iter().rev()),
190            _ => {}
191        }
192    }
193    if answers.is_empty() {
194        outcome(ctx, id, operation, "model_silent");
195        anyhow::bail!("the handler gave no answer");
196    }
197    Ok(answers)
198}
199
200struct Conn<'a> {
201    shared: &'a Shared,
202    id: ConnectionId,
203    r: tokio::io::ReadHalf<TcpStream>,
204    w: tokio::io::WriteHalf<TcpStream>,
205    buf: Vec<u8>,
206}
207
208impl Conn<'_> {
209    async fn write(&mut self, chunks: &[Vec<u8>]) -> Result<()> {
210        for c in chunks.iter().filter(|c| !c.is_empty()) {
211            self.w.write_all(c).await?;
212            self.shared
213                .ctx
214                .state
215                .update_connection_stats(
216                    self.shared.ctx.server_id,
217                    self.id,
218                    None,
219                    Some(c.len() as u64),
220                    None,
221                    Some(1),
222                )
223                .await;
224        }
225        Ok(())
226    }
227
228    /// The next whole message, skipping garbled input; `None` on EOF.
229    async fn next(&mut self) -> Result<Option<(String, Message)>> {
230        loop {
231            match codec::frame(&self.buf) {
232                Frame::Message {
233                    begin,
234                    message,
235                    consumed,
236                } => {
237                    self.buf.drain(..consumed);
238                    self.shared
239                        .ctx
240                        .state
241                        .update_connection_stats(
242                            self.shared.ctx.server_id,
243                            self.id,
244                            Some(consumed as u64),
245                            None,
246                            Some(1),
247                            None,
248                        )
249                        .await;
250                    return Ok(Some((begin, message)));
251                }
252                Frame::Garbled { consumed, reason } => {
253                    self.buf.drain(..consumed);
254                    Log::new(Some(&self.shared.ctx.status_tx)).warn(format!(
255                        "FIX connection {}: ignored garbled input ({reason})",
256                        self.id
257                    ));
258                }
259                Frame::Incomplete => {
260                    anyhow::ensure!(
261                        self.buf.len() <= codec::MAX_BODY + 64,
262                        "unframed input over {} bytes",
263                        codec::MAX_BODY
264                    );
265                    let mut chunk = [0u8; 8192];
266                    let n = self.r.read(&mut chunk).await?;
267                    if n == 0 {
268                        return Ok(None);
269                    }
270                    self.buf.extend_from_slice(&chunk[..n]);
271                }
272            }
273        }
274    }
275}
276
277async fn connection(shared: &Shared, id: ConnectionId, stream: TcpStream) -> Result<()> {
278    let ctx = &shared.ctx;
279    let (r, w) = tokio::io::split(stream);
280    let mut c = Conn {
281        shared,
282        id,
283        r,
284        w,
285        buf: Vec::new(),
286    };
287    // The Logon phase: the first message must be a valid Logon (FIX 4.4 vol 2, "Logon").
288    let Ok(first) = tokio::time::timeout(shared.logon_timeout, c.next()).await else {
289        outcome(ctx, id, "logon", "protocol_refusal");
290        return Ok(());
291    };
292    let Some((begin, logon)) = first? else {
293        return Ok(());
294    };
295    if logon.msg_type() != "A" || begin != shared.begin {
296        outcome(ctx, id, "logon", "protocol_refusal");
297        anyhow::bail!(
298            "the first message was {} {}, not a {} Logon; disconnected",
299            begin,
300            logon.msg_type(),
301            shared.begin
302        );
303    }
304    let peer = logon.get(49).unwrap_or_default().to_owned();
305    let heartbeat = logon
306        .get(108)
307        .and_then(|v| v.parse::<u64>().ok())
308        .filter(|h| *h <= 3600);
309    let refusal = if !comp_id_ok(&peer) {
310        Some("SenderCompID missing or invalid")
311    } else if logon.get(56) != Some(shared.comp_id.as_str()) {
312        Some("TargetCompID is not this acceptor")
313    } else if heartbeat.is_none() {
314        Some("HeartBtInt must be 0 to 3600")
315    } else if logon.get(98).unwrap_or("0") != "0" {
316        Some("only EncryptMethod 0 is supported")
317    } else if logon.seq().is_none() {
318        Some("MsgSeqNum missing")
319    } else {
320        None
321    };
322    let hb = heartbeat.unwrap_or(30);
323    let mut session = Session::new(
324        &shared.begin,
325        &shared.comp_id,
326        if peer.is_empty() { "UNKNOWN" } else { &peer },
327        if hb == 0 {
328            NO_HEARTBEAT
329        } else {
330            Duration::from_secs(hb)
331        },
332    );
333    if let Some(text) = refusal {
334        outcome(ctx, id, "logon", "protocol_refusal");
335        let bytes = session.logout(Some(text));
336        c.write(&[bytes]).await?;
337        return Ok(());
338    }
339    let reset = logon.get(141) == Some("Y");
340    let event = Event::new(
341        &actions::LOGON_EVENT,
342        json!({"sender_comp_id": peer, "target_comp_id": shared.comp_id, "heartbeat_secs": hb, "reset_seq_num": reset, "username": logon.get(553), "password": logon.get(554)}),
343    );
344    match ask(shared, id, event, "logon")
345        .await
346        .map(|mut a| a.remove(0))
347    {
348        Ok(a) if a["type"] == "fix_accept_logon" => outcome(ctx, id, "logon", "model_answer"),
349        Ok(a) if a["type"] == "fix_reject_logon" => {
350            outcome(ctx, id, "logon", "model_reject");
351            let bytes = session.logout(a["text"].as_str());
352            c.write(&[bytes]).await?;
353            return Ok(());
354        }
355        other => {
356            if let Ok(a) = other {
357                outcome(ctx, id, "logon", "fail_closed_invalid_reply");
358                Log::new(Some(&ctx.status_tx)).warn(format!(
359                    "FIX connection {id}: {} does not answer a Logon",
360                    a["type"]
361                ));
362            }
363            let bytes = session.logout(Some("logon cannot be processed now"));
364            c.write(&[bytes]).await?;
365            return Ok(());
366        }
367    }
368    session.accept_logon(logon.seq().unwrap_or(1));
369    let mut reply = vec![(98, "0".to_owned()), (108, hb.to_string())];
370    if reset {
371        reply.push((141, "Y".into()));
372    }
373    let bytes = session.encode("A", &reply)?;
374    c.write(&[bytes]).await?;
375    let mut commands =
376        crate::server::peer_support::register_peer_channel(&ctx.state, ctx.server_id, id.as_u32())
377            .await;
378    session_loop(&mut c, &mut session, &mut commands).await
379}
380
381async fn session_loop(
382    c: &mut Conn<'_>,
383    session: &mut Session,
384    commands: &mut mpsc::Receiver<ClientCommand>,
385) -> Result<()> {
386    let ctx = &c.shared.ctx;
387    let id = c.id;
388    let mut tick = tokio::time::interval(Duration::from_secs(1));
389    let mut closing: Option<Instant> = None;
390    loop {
391        if closing.is_some_and(|t| t.elapsed() >= LOGOUT_WAIT) {
392            return Ok(());
393        }
394        enum Wake {
395            Message(Option<Message>),
396            Tick,
397            Command(Option<ClientCommand>),
398        }
399        let wake = tokio::select! {
400            m = c.next() => Wake::Message(m?.map(|(_, m)| m)),
401            _ = tick.tick() => Wake::Tick,
402            cmd = commands.recv(), if closing.is_none() => Wake::Command(cmd),
403        };
404        let message = match wake {
405            Wake::Message(Some(m)) => m,
406            Wake::Message(None) => return Ok(()),
407            Wake::Tick => {
408                let (send, close) = session.tick();
409                c.write(&send).await?;
410                if let Some(reason) = close {
411                    Log::new(Some(&ctx.status_tx)).warn(format!("FIX connection {id}: {reason}"));
412                    return Ok(());
413                }
414                continue;
415            }
416            Wake::Command(None) => continue,
417            Wake::Command(Some(cmd)) => {
418                let action = cmd.action.clone();
419                let outcome = injected(c, session, &action).await;
420                if session.logout_sent {
421                    closing.get_or_insert_with(Instant::now);
422                }
423                ctx.state
424                    .record_access_log(
425                        crate::state::AccessLogOwner::Server(ctx.server_id.as_u32()),
426                        "FIX",
427                        Some(id.as_u32()),
428                        "injected_action",
429                        json!({"type": action["type"], "msg_type": action["msg_type"]}),
430                        vec![serde_json::to_value(&outcome).unwrap_or(Value::Null)],
431                    )
432                    .await;
433                let _ = cmd.reply_tx.send(Ok(outcome));
434                continue;
435            }
436        };
437        let step = session.on_message(message);
438        c.write(&step.send).await?;
439        if let Some(reason) = step.close {
440            Log::new(Some(&ctx.status_tx)).info(format!("FIX connection {id}: {reason}"));
441            return Ok(());
442        }
443        let Some(app) = step.app else { continue };
444        if closing.is_some() {
445            continue;
446        }
447        let msg_type = app.msg_type().to_owned();
448        let seq = app.seq().unwrap_or(0);
449        let fields: Vec<Value> = session::body(&app)
450            .into_iter()
451            .map(|(tag, value)| json!({"tag": tag, "name": dict::field_name(tag), "value": value}))
452            .collect();
453        let event = Event::new(
454            &actions::MESSAGE_EVENT,
455            json!({"msg_type": msg_type, "msg_type_name": dict::message_name(&msg_type), "seq": seq, "sender_comp_id": session.target, "fields": fields}),
456        );
457        match ask(c.shared, id, event, "message").await {
458            Ok(answers) => {
459                outcome(
460                    ctx,
461                    id,
462                    "message",
463                    if answers.iter().any(|a| a["type"] == "fix_reject") {
464                        "model_reject"
465                    } else {
466                        "model_answer"
467                    },
468                );
469                for a in answers {
470                    apply(c, session, &a, seq, &msg_type).await?;
471                }
472                if session.logout_sent {
473                    closing.get_or_insert_with(Instant::now);
474                }
475            }
476            Err(e) => {
477                let text = crate::utils::WireFailure::classify(&e).text();
478                let bytes = business_reject(session, seq, &msg_type, 4, text)?;
479                c.write(&[bytes]).await?;
480            }
481        }
482    }
483}
484
485fn business_reject(
486    session: &mut Session,
487    ref_seq: u32,
488    ref_type: &str,
489    reason: u32,
490    text: &str,
491) -> Result<Vec<u8>> {
492    session.encode(
493        "j",
494        &[
495            (45, ref_seq.to_string()),
496            (372, ref_type.to_owned()),
497            (380, reason.to_string()),
498            (58, text.to_owned()),
499        ],
500    )
501}
502
503/// Apply one handler answer to an application message.
504async fn apply(
505    c: &mut Conn<'_>,
506    session: &mut Session,
507    a: &Value,
508    ref_seq: u32,
509    ref_type: &str,
510) -> Result<()> {
511    let bytes = match a["type"].as_str() {
512        Some("fix_send") => {
513            let msg_type = actions::app_msg_type(a)?;
514            session.encode(&msg_type, &actions::body_fields(a)?)?
515        }
516        Some("fix_reject") => {
517            let reason = actions::BUSINESS_REJECT
518                .iter()
519                .find(|(n, _)| a["reason"] == *n)
520                .map(|(_, c)| *c)
521                .unwrap_or(0);
522            business_reject(
523                session,
524                ref_seq,
525                ref_type,
526                reason,
527                a["text"].as_str().unwrap_or_default(),
528            )?
529        }
530        Some("fix_logout") => session.logout(a["text"].as_str()),
531        _ => return Ok(()),
532    };
533    c.write(&[bytes]).await
534}
535
536/// An action injected from the dashboard or MCP: fix_send, fix_logout or disconnect.
537async fn injected(c: &mut Conn<'_>, session: &mut Session, a: &Value) -> ClientSendOutcome {
538    let checked = match a["type"].as_str() {
539        Some("fix_send" | "fix_logout") => actions::check_answer(a),
540        Some("disconnect") => Ok(()),
541        _ => Err(anyhow::anyhow!(
542            "a FIX session accepts fix_send, fix_logout or disconnect"
543        )),
544    };
545    if let Err(e) = checked {
546        return ClientSendOutcome::Rejected {
547            error: e.to_string(),
548        };
549    }
550    let bytes = match a["type"].as_str() {
551        Some("fix_send") => match actions::app_msg_type(a)
552            .and_then(|t| session.encode(&t, &actions::body_fields(a)?))
553        {
554            Ok(b) => b,
555            Err(e) => {
556                return ClientSendOutcome::Rejected {
557                    error: e.to_string(),
558                }
559            }
560        },
561        Some("fix_logout") => session.logout(a["text"].as_str()),
562        _ => session.logout(None),
563    };
564    let n = bytes.len();
565    match c.write(&[bytes]).await {
566        Ok(()) if a["type"] == "disconnect" => ClientSendOutcome::Disconnected,
567        Ok(()) => ClientSendOutcome::Sent { bytes_sent: n },
568        Err(e) => ClientSendOutcome::Rejected {
569            error: e.to_string(),
570        },
571    }
572}
netget/server/fix/mod.rs

netget/server/fix/
mod.rs
