forked from
microcosm.blue/microcosm-rs
Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
handle jetstream connection issues without crashing
+36
-5
+1
link_aggregator/readme.md
+1
link_aggregator/readme.md
···
117
117
- [ ] put delete-account tasks into a separate (persisted?) task queue for the writer so it can work on them incrementally.
118
118
- [ ] jetstream: connect retry: only reset counter after some *time* has passed.
119
119
- [x] either count or estimate the total number of links added (distinct from link targets)
120
+
- [x] jetstream: don't crash on connection refused (retry * backoff)
120
121
121
122
cache
122
123
- [ ] set api response headers
+35
-5
link_aggregator/src/consumer/jetstream.rs
+35
-5
link_aggregator/src/consumer/jetstream.rs
···
85
85
let host = req.uri().host().expect("jetstream request uri has a host");
86
86
let port = req.uri().port().map(|p| p.as_u16()).unwrap_or(443);
87
87
let dest = format!("{host}:{port}");
88
-
let addr = dest
89
-
.to_socket_addrs()?
90
-
.next()
91
-
.expect("can resolve the jetstream address"); // TODO ugh
92
-
let tcp_stream = std::net::TcpStream::connect_timeout(&addr, time::Duration::from_secs(8))?; // TODO: handle
88
+
let addr = match dest.to_socket_addrs().map(|mut d| d.next()) {
89
+
Ok(Some(a)) => a,
90
+
Ok(None) => {
91
+
eprintln!(
92
+
"jetstream: could not resolve an address for {dest:?}. retrying after a bit?"
93
+
);
94
+
thread::sleep(time::Duration::from_secs(15));
95
+
continue;
96
+
}
97
+
Err(e) => {
98
+
eprintln!("jetstream failed to resolve address {dest:?}: {e:?} waiting and then retrying...");
99
+
thread::sleep(time::Duration::from_secs(3));
100
+
continue;
101
+
}
102
+
};
103
+
let tcp_stream = match std::net::TcpStream::connect_timeout(
104
+
&addr,
105
+
time::Duration::from_secs(8),
106
+
) {
107
+
Ok(s) => s,
108
+
Err(e) => {
109
+
eprintln!(
110
+
"jetstream failed to make tcp connection: {e:?}. (todo: clean up retry logic)"
111
+
);
112
+
connect_retries += 1;
113
+
if connect_retries >= 7 {
114
+
eprintln!("jetstream: no more connect retries, breaking out.");
115
+
break;
116
+
}
117
+
let backoff = time::Duration::from_secs(connect_retries.try_into().unwrap());
118
+
eprintln!("jetstream tcp failed to connect: {e:?}. backing off {backoff:?} before retrying...");
119
+
thread::sleep(backoff);
120
+
continue;
121
+
}
122
+
};
93
123
tcp_stream.set_read_timeout(Some(time::Duration::from_secs(4)))?;
94
124
tcp_stream.set_write_timeout(Some(time::Duration::from_secs(4)))?;
95
125