Fix GC crash from pointer-typed session handles

.changeset/bindings-gc-crash.md

@@ -0,0 +1,5 @@
+---
+"@kurrent/gaffer": patch
+---
+
+The Go runtime bindings no longer risk a rare fatal crash (`invalid pointer found on stack`) under GC pressure. The runtime's integer session handles could appear in pointer-typed stack slots during FFI calls; if the GC moved the stack while a callback was running mid-call, the process aborted. Handles now stay integer-typed end-to-end on the Go side, with all casting done in C shims. The crash could hit anything embedding the bindings, including the CLI.

bindings/go/callbacks.go

@@ -2,20 +2,61 @@ package gafferruntime
 
 /*
 #include "gaffer.h"
+#include <stdint.h>
 
-// Forward declarations for Go callback trampolines
-extern void goEmitCallback(const char* streamId, const char* eventType, const char* data, const char* metadata, int isJson, int isLink, void* userData);
-extern void goLogCallback(const char* message, void* userData);
-extern void goDiagnosticCallback(const char* code, const char* message, int severity, void* userData);
-extern void goStateChangedCallback(const char* partition, const char* stateJson, void* userData);
-extern void goBreakCallback(const char* reason, const char* source, int line, int column, void* userData);
+// Forward declarations for Go callback trampolines. user_data is declared
+// uintptr_t to match the //export signatures: it carries the session handle,
+// a small integer that must never appear pointer-typed on the Go side (see
+// the shim comment in gaffer.go's preamble).
+extern void goEmitCallback(const char* streamId, const char* eventType, const char* data, const char* metadata, int isJson, int isLink, uintptr_t userData);
+extern void goLogCallback(const char* message, uintptr_t userData);
+extern void goDiagnosticCallback(const char* code, const char* message, int severity, uintptr_t userData);
+extern void goStateChangedCallback(const char* partition, const char* stateJson, uintptr_t userData);
+extern void goBreakCallback(const char* reason, const char* source, int line, int column, uintptr_t userData);
+
+// Thunks matching the gaffer_*_cb typedefs exactly. The void* -> uintptr_t
+// conversion happens here on the user_data VALUE; casting the Go trampolines
+// (uintptr_t last parameter) to the void*-taking typedefs instead would call
+// through an incompatible function-pointer type, which is undefined behavior.
+static void go_emit_thunk(const char* streamId, const char* eventType, const char* data, const char* metadata, int isJson, int isLink, void* userData) {
+	goEmitCallback(streamId, eventType, data, metadata, isJson, isLink, (uintptr_t)userData);
+}
+static void go_log_thunk(const char* message, void* userData) {
+	goLogCallback(message, (uintptr_t)userData);
+}
+static void go_diagnostic_thunk(const char* code, const char* message, int severity, void* userData) {
+	goDiagnosticCallback(code, message, severity, (uintptr_t)userData);
+}
+static void go_state_changed_thunk(const char* partition, const char* stateJson, void* userData) {
+	goStateChangedCallback(partition, stateJson, (uintptr_t)userData);
+}
+static void go_break_thunk(const char* reason, const char* source, int line, int column, void* userData) {
+	goBreakCallback(reason, source, line, column, (uintptr_t)userData);
+}
+
+// Registration shims: cast the integer handle to gaffer_session* (and back
+// into the user_data slot) in C, so no fake pointer transits Go stack slots.
+static void go_on_emit(uintptr_t s) {
+	gaffer_on_emit((gaffer_session*)s, go_emit_thunk, (void*)s);
+}
+static void go_on_log(uintptr_t s) {
+	gaffer_on_log((gaffer_session*)s, go_log_thunk, (void*)s);
+}
+static void go_on_diagnostic(uintptr_t s) {
+	gaffer_on_diagnostic((gaffer_session*)s, go_diagnostic_thunk, (void*)s);
+}
+static void go_on_state_changed(uintptr_t s) {
+	gaffer_on_state_changed((gaffer_session*)s, go_state_changed_thunk, (void*)s);
+}
+static void go_on_break(uintptr_t s) {
+	gaffer_on_break((gaffer_session*)s, go_break_thunk, (void*)s);
+}
 */
 import "C"
 
 import (
 	"sync"
 	"sync/atomic"
-	"unsafe"
 )
 
 // Callback function types.
@@ -27,7 +68,7 @@ type (
 	BreakCallback        func(info BreakInfo)
 )
 
-// Global callback registry keyed by session pointer.
+// Global callback registry keyed by session handle.
 var (
 	callbackMu          sync.RWMutex
 	emitCallbacks       = make(map[uintptr]EmitCallback)
@@ -37,52 +78,42 @@ var (
 	breakCallbacks      = make(map[uintptr]BreakCallback)
 )
 
-func sessionKey(session *C.gaffer_session) uintptr {
-	return uintptr(unsafe.Pointer(session))
-}
-
-func sessionOnEmit(session *C.gaffer_session, cb EmitCallback) {
-	key := sessionKey(session)
+func sessionOnEmit(handle uintptr, cb EmitCallback) {
 	callbackMu.Lock()
-	emitCallbacks[key] = cb
+	emitCallbacks[handle] = cb
 	callbackMu.Unlock()
-	C.gaffer_on_emit(session, (*[0]byte)(C.goEmitCallback), unsafe.Pointer(session))
+	C.go_on_emit(C.uintptr_t(handle))
 }
 
-func sessionOnLog(session *C.gaffer_session, cb LogCallback) {
-	key := sessionKey(session)
+func sessionOnLog(handle uintptr, cb LogCallback) {
 	callbackMu.Lock()
-	logCallbacks[key] = cb
+	logCallbacks[handle] = cb
 	callbackMu.Unlock()
-	C.gaffer_on_log(session, (*[0]byte)(C.goLogCallback), unsafe.Pointer(session))
+	C.go_on_log(C.uintptr_t(handle))
 }
 
-func sessionOnDiagnostic(session *C.gaffer_session, cb DiagnosticCallback) {
-	key := sessionKey(session)
+func sessionOnDiagnostic(handle uintptr, cb DiagnosticCallback) {
 	callbackMu.Lock()
-	diagnosticCallbacks[key] = cb
+	diagnosticCallbacks[handle] = cb
 	callbackMu.Unlock()
-	C.gaffer_on_diagnostic(session, (*[0]byte)(C.goDiagnosticCallback), unsafe.Pointer(session))
+	C.go_on_diagnostic(C.uintptr_t(handle))
 }
 
-func sessionOnStateChanged(session *C.gaffer_session, cb StateChangedCallback) {
-	key := sessionKey(session)
+func sessionOnStateChanged(handle uintptr, cb StateChangedCallback) {
 	callbackMu.Lock()
-	changedCallbacks[key] = cb
+	changedCallbacks[handle] = cb
 	callbackMu.Unlock()
-	C.gaffer_on_state_changed(session, (*[0]byte)(C.goStateChangedCallback), unsafe.Pointer(session))
+	C.go_on_state_changed(C.uintptr_t(handle))
 }
 
-func sessionOnBreak(session *C.gaffer_session, cb BreakCallback) {
-	key := sessionKey(session)
+func sessionOnBreak(handle uintptr, cb BreakCallback) {
 	callbackMu.Lock()
-	breakCallbacks[key] = cb
+	breakCallbacks[handle] = cb
 	callbackMu.Unlock()
-	C.gaffer_on_break(session, (*[0]byte)(C.goBreakCallback), unsafe.Pointer(session))
+	C.go_on_break(C.uintptr_t(handle))
 }
 
-func cleanupCallbacks(session *C.gaffer_session) {
-	key := sessionKey(session)
+func cleanupCallbacks(key uintptr) {
 	callbackMu.Lock()
 	delete(emitCallbacks, key)
 	delete(logCallbacks, key)