Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
jetstream: set tcp timeouts
for the moment it's arbitrarily hard-coded at 8s for connect, and 4s for both read and write.
i'm not really sure what the problem would be for doing it this way, but according to the tungstenite author, [i'm probably doing it wrong](https://github.com/snapview/tungstenite-rs/issues/252#issuecomment-984144901)
mayyyybe using tungstunite-rs is too low level and i really should embrace #[tokio:main]. probably. but i'm going to hold out a minute longer.
+19
-3
+3
-1
cozy-setup (move to another repo).md
+3
-1
cozy-setup (move to another repo).md
···
164
164
- [x] configure caddy rate-limit
165
165
- [ ] configure caddy to use a health check (once it's added)
166
166
- [ ] configure caddy to only expose cache metrics to tailnet :/
167
-
- [ ] make some grafana dashboards
167
+
- [x] make some grafana dashboards
168
+
- [ ] raspi: mount /dev/sda on boot
169
+
- [ ] raspi: run link_aggregator via systemd so it starts on startup (and restarts?)
168
170
+3
-1
link_aggregator/readme.md
+3
-1
link_aggregator/readme.md
···
81
81
- [ ] get count + most recent dids
82
82
- [ ] get count with any dids from provided set
83
83
- [~] write this readme
84
-
- [ ] fix it sometimes getting stuck
84
+
- [?] fix it sometimes getting stuck
85
85
- seems to unstick in my possibly-different repro (letting laptop fall asleep) after a bit.
86
86
- [ ] add a detection for no new links coming in after some period
87
+
- [x] add tcp connect, read, and write timeouts 🤞
87
88
- [x] handle jetstream restart: don't miss events (currently sketch: rewinds cursor by 1us so we will always double-count at least one event)
88
89
- [x] especially: figure out what the risk is to rotating to another jetstream server in terms of gap/overlap from a different jetstream instance's cursor (follow up separately)
89
90
- [x] jetstream: don't rotate servers, explicitly pass via cli
···
112
113
- [ ] add a health check endpoint
113
114
- [ ] add seq numbers to metrics
114
115
- [ ] persist the jetstream server url, error if started with a different one (maybe with --switch-streams or something)
116
+
- [ ] put delete-account tasks into a separate (persisted?) task queue for the writer so it can work on them incrementally.
115
117
116
118
data fixes
117
119
- [ ] add rkey to linkers 🤦♀️
+13
-1
link_aggregator/src/consumer/jetstream.rs
+13
-1
link_aggregator/src/consumer/jetstream.rs
···
3
3
counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram, Unit,
4
4
};
5
5
use std::io::{Cursor, Read};
6
+
use std::net::ToSocketAddrs;
6
7
use std::thread;
7
8
use std::time;
8
9
use tinyjson::JsonValue;
···
81
82
let ua = format!("ucosm/link aggregator v{}", env!("CARGO_PKG_VERSION"));
82
83
req.headers_mut().insert("user-agent", ua.parse()?);
83
84
85
+
let host = req.uri().host().expect("jetstream request uri has a host");
86
+
let port = req.uri().port().map(|p| p.as_u16()).unwrap_or(443);
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
93
+
tcp_stream.set_read_timeout(Some(time::Duration::from_secs(4)))?;
94
+
tcp_stream.set_write_timeout(Some(time::Duration::from_secs(4)))?;
95
+
84
96
counter!("jetstream_connect", "url" => stream.clone(), "is_retry" => (connect_retries > 0).to_string()).increment(1);
85
97
println!("jetstream connecting, attempt #{connect_retries}, {stream_url:?} with user-agent: {ua:?}");
86
-
let mut socket = match tungstenite::connect(req) {
98
+
let mut socket = match tungstenite::client_tls(req, tcp_stream) {
87
99
Ok((socket, _)) => {
88
100
println!("jetstream connected.");
89
101
connect_retries = 0;