Skip to content

Commit 13390c8

Browse files
alistairjevansclaude
alistairjevans
and
claude
authored
Make unix socket takeover atomic and guard cleanup by inode (#8)
After a full docker restart, a replacement ingester can bind its unix socket while the old container is still draining; the old server's stop() then unlinked the path by name, deleting the new server's socket. The new server keeps listening on an unlinked inode and never recovers (BetterStackHQ Linear T-8292). - start() now binds to a unique temp path and atomically renames it over the target, so the path always points at a live socket and takeover of a still-bound path is a single step. - stop() records the bound socket file's (dev, ino) and only unlinks the path if it still matches, so an older generation never removes a newer generation's socket. Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 1a8bc56 commit 13390c8

2 files changed

Lines changed: 116 additions & 34 deletions

File tree

‎ext/hyper_ruby/src/lib.rs‎

Lines changed: 45 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ use tokio::io::{AsyncRead, AsyncWrite};
1616

1717
use std::cell::RefCell;
1818
use std::net::SocketAddr;
19+
use std::os::unix::fs::MetadataExt;
1920
use std::sync::atomic::{AtomicU64, Ordering};
2021

2122
use tokio::net::{TcpListener, UnixListener};
@@ -113,6 +114,9 @@ struct Server {
113114
runtime: RefCell<Option<Arc<tokio::runtime::Runtime>>>,
114115
shutdown: RefCell<Option<broadcast::Sender<()>>>,
115116
total_connections: Arc<AtomicU64>,
117+
// (dev, ino) of the Unix socket file this server bound, so stop() only removes
118+
// the file if a replacement server hasn't taken over the path in the meantime.
119+
socket_ident: RefCell<Option<(u64, u64)>>,
116120
}
117121

118122
impl Server {
@@ -126,6 +130,7 @@ impl Server {
126130
runtime: RefCell::new(None),
127131
shutdown: RefCell::new(None),
128132
total_connections: Arc::new(AtomicU64::new(0)),
133+
socket_ident: RefCell::new(None),
129134
}
130135
}
131136

@@ -310,32 +315,40 @@ impl Server {
310315
// Create the listener with proper error handling
311316
let listener = if config.bind_address.starts_with("unix:") {
312317
let path = config.bind_address.trim_start_matches("unix:");
313-
314-
// Check if the socket file already exists and try to delete it
315-
if std::path::Path::new(path).exists() {
316-
debug!("Unix socket file {} already exists, attempting to remove it", path);
317-
match std::fs::remove_file(path) {
318-
Ok(_) => debug!("Successfully removed existing socket file"),
319-
Err(e) => {
320-
error!("Failed to remove existing Unix socket file {}: {}", path, e);
321-
return Err(MagnusError::new(
322-
magnus::exception::runtime_error(),
323-
format!("Failed to remove existing Unix socket file {}: {}", path, e)
324-
));
325-
}
326-
}
327-
}
328-
329-
match UnixListener::bind(path) {
330-
Ok(listener) => Listener::Unix(listener),
318+
319+
// Bind to a unique temp path and atomically rename over the target, so the
320+
// path always points at a live socket even when a replacement server takes
321+
// over a path an older, still-draining server bound.
322+
static SOCKET_TMP_SEQ: AtomicU64 = AtomicU64::new(0);
323+
let tmp_path = format!("{}.{}.{}.tmp", path, std::process::id(), SOCKET_TMP_SEQ.fetch_add(1, Ordering::Relaxed));
324+
325+
let listener = match UnixListener::bind(&tmp_path) {
326+
Ok(listener) => listener,
331327
Err(e) => {
332-
error!("Failed to bind to Unix socket {}: {}", path, e);
328+
error!("Failed to bind to Unix socket {}: {}", tmp_path, e);
333329
return Err(MagnusError::new(
334330
magnus::exception::runtime_error(),
335-
format!("Failed to bind to Unix socket {}: {}", path, e)
331+
format!("Failed to bind to Unix socket {}: {}", tmp_path, e)
336332
));
337333
}
334+
};
335+
336+
// The socket file's identity survives the rename; stop() compares against it
337+
// so an older server generation never unlinks a newer generation's socket.
338+
let ident = std::fs::symlink_metadata(&tmp_path).ok().map(|m| (m.dev(), m.ino()));
339+
340+
if let Err(e) = std::fs::rename(&tmp_path, path) {
341+
let _ = std::fs::remove_file(&tmp_path);
342+
error!("Failed to install Unix socket file {}: {}", path, e);
343+
return Err(MagnusError::new(
344+
magnus::exception::runtime_error(),
345+
format!("Failed to install Unix socket file {}: {}", path, e)
346+
));
338347
}
348+
349+
*self.socket_ident.borrow_mut() = ident;
350+
351+
Listener::Unix(listener)
339352
} else {
340353
match config.bind_address.parse::<SocketAddr>() {
341354
Ok(addr) => {
@@ -474,9 +487,18 @@ impl Server {
474487
let bind_address = self.config.borrow().bind_address.clone();
475488
if bind_address.starts_with("unix:") {
476489
let path = bind_address.trim_start_matches("unix:");
477-
std::fs::remove_file(path).unwrap_or_else(|e| {
478-
warn!("Failed to remove socket file: {:?}", e);
479-
});
490+
// Only remove the socket file if it's still the one this server bound; a
491+
// replacement server may have taken over the path while we were draining.
492+
match (self.socket_ident.borrow_mut().take(), std::fs::symlink_metadata(path)) {
493+
(Some((dev, ino)), Ok(meta)) if (meta.dev(), meta.ino()) == (dev, ino) => {
494+
std::fs::remove_file(path).unwrap_or_else(|e| {
495+
warn!("Failed to remove socket file: {:?}", e);
496+
});
497+
}
498+
(Some(_), Ok(_)) => info!("Socket file {} was replaced by another server; leaving it in place", path),
499+
(Some(_), Err(_)) => debug!("Socket file {} already removed", path),
500+
(None, _) => {}
501+
}
480502
}
481503

482504
Ok(())

‎test/test_http.rb‎

Lines changed: 71 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -153,6 +153,65 @@ def test_unix_socket_cleanup
153153
end
154154
end
155155

156+
def test_unix_socket_takeover_preserves_new_generation_socket
157+
# A replacement server (e.g. a redeployed container) can bind the same path while the
158+
# old server is still draining; the old server's stop must not delete the new socket.
159+
socket_path = "/tmp/hyper_ruby_test_takeover.sock"
160+
File.unlink(socket_path) if File.exist?(socket_path)
161+
162+
old_server = HyperRuby::Server.new
163+
old_server.configure({ bind_address: "unix:#{socket_path}" })
164+
old_server.start
165+
166+
new_server = HyperRuby::Server.new
167+
new_server.configure({ bind_address: "unix:#{socket_path}" })
168+
new_server.start
169+
170+
workers = 1.times.map do
171+
Thread.new do
172+
new_server.run_worker { |request| handler_simple(request) }
173+
end
174+
end
175+
176+
old_server.stop
177+
old_server = nil
178+
assert File.exist?(socket_path), "old server's stop must not remove the new server's socket"
179+
180+
client = HTTPX.with(transport: "unix", addresses: [socket_path], origin: "http://host")
181+
response = client.get("/")
182+
assert_equal 200, response.status
183+
184+
new_server.stop
185+
workers.each(&:join)
186+
workers = nil
187+
refute File.exist?(socket_path), "new server's stop should remove the socket it owns"
188+
ensure
189+
old_server.stop if old_server
190+
new_server.stop if new_server && workers
191+
workers&.each(&:join)
192+
File.unlink(socket_path) if File.exist?(socket_path)
193+
end
194+
195+
def test_unix_socket_stop_leaves_foreign_file
196+
# If something else has replaced our socket file, stop must leave it alone.
197+
socket_path = "/tmp/hyper_ruby_test_foreign.sock"
198+
File.unlink(socket_path) if File.exist?(socket_path)
199+
200+
server = HyperRuby::Server.new
201+
server.configure({ bind_address: "unix:#{socket_path}" })
202+
server.start
203+
204+
File.unlink(socket_path)
205+
FileUtils.touch(socket_path)
206+
207+
server.stop
208+
server = nil
209+
assert File.exist?(socket_path), "stop must not remove a file it did not bind"
210+
ensure
211+
server.stop if server
212+
File.unlink(socket_path) if File.exist?(socket_path)
213+
end
214+
156215
# This test requires root permissions to create a file that can't be deleted.
157216
# Skip it unless we're running with proper permissions.
158217
def test_unix_socket_undeletable
@@ -171,13 +230,13 @@ def test_unix_socket_undeletable
171230
server = HyperRuby::Server.new
172231
server.configure({ bind_address: "unix:#{socket_path}" })
173232

174-
# This should raise an exception about not being able to remove the file
233+
# This should raise an exception about not being able to install the socket file
175234
error = assert_raises(RuntimeError) do
176235
server.start
177236
end
178-
237+
179238
# Verify the error message
180-
assert_match(/Failed to remove existing Unix socket file/, error.message)
239+
assert_match(/Failed to install Unix socket file/, error.message)
181240
ensure
182241
# Clean up with sudo
183242
system("sudo rm -f #{socket_path}") if File.exist?(socket_path)
@@ -201,14 +260,15 @@ def test_unix_socket_directory_error
201260
server.start
202261
end
203262

204-
# The error is from trying to remove the directory, not from binding
205-
assert_match(/Failed to remove existing Unix socket file/, error.message)
206-
207-
# It should include something about "Operation not permitted" or similar
208-
assert(error.message.include?("Operation not permitted") ||
209-
error.message.include?("Permission denied") ||
210-
error.message.include?("not a socket"),
211-
"Error should indicate issue with removing directory: #{error.message}")
263+
# The error is from renaming the bound socket over the directory, not from binding
264+
assert_match(/Failed to install Unix socket file/, error.message)
265+
266+
# It should include something about the target being a directory or similar
267+
assert(error.message.include?("Is a directory") ||
268+
error.message.include?("Operation not permitted") ||
269+
error.message.include?("Permission denied") ||
270+
error.message.include?("not a socket"),
271+
"Error should indicate issue with replacing directory: #{error.message}")
212272
ensure
213273
# Clean up
214274
FileUtils.rm_rf(socket_dir) if Dir.exist?(socket_dir)

0 commit comments

Comments
 (0)