Compare commits

...

1 Commits

Author SHA1 Message Date
Hein 16cc7d350e fix(hooks): implement retry logic for hook registry locks
Tests / Unit Tests (push) Failing after 10s
Tests / Integration Tests (push) Failing after 13s
Build , Vet Test, and Lint / Run Vet Tests (1.23.x) (push) Successful in 1m5s
Build , Vet Test, and Lint / Run Vet Tests (1.24.x) (push) Successful in 1m3s
Build , Vet Test, and Lint / Build (push) Successful in 1m7s
Build , Vet Test, and Lint / Lint Code (push) Successful in 1m48s
* Add tryLock and tryRLock methods to manage mutex access
* Update Register, Clear, and Execute methods to handle locked state
* Log errors when registry operations fail due to locking
2026-08-04 17:55:36 +02:00
6 changed files with 488 additions and 233 deletions
+81 -6
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"reflect"
"sync"
"time"
)
// ModelRules defines the permissions and security settings for a model
@@ -59,12 +60,44 @@ func NewModelRegistry() *DefaultModelRegistry {
}
}
// lockRetryAttempts/lockRetryDelay bound how long the try-lock helpers below
// will spin before giving up, so a contended registriesMutex can never hang
// a caller of GetDefaultRegistry/SetDefaultRegistry.
const (
lockRetryAttempts = 20
lockRetryDelay = 1 * time.Millisecond
)
// GetDefaultRegistry returns the current default registry. It uses a
// bounded TryRLock instead of a blocking RLock so it can never hang;
// if the lock can't be acquired in time it falls back to the last known
// value without synchronization.
func GetDefaultRegistry() *DefaultModelRegistry {
for i := 0; i < lockRetryAttempts; i++ {
if registriesMutex.TryRLock() {
defer registriesMutex.RUnlock()
return defaultRegistry
}
time.Sleep(lockRetryDelay)
}
return defaultRegistry
}
// SetDefaultRegistry replaces the default registry. It uses a bounded
// TryLock instead of a blocking Lock so it can never hang; if the lock
// can't be acquired in time the call is a no-op.
func SetDefaultRegistry(registry *DefaultModelRegistry) {
registriesMutex.Lock()
acquired := false
for i := 0; i < lockRetryAttempts; i++ {
if registriesMutex.TryLock() {
acquired = true
break
}
time.Sleep(lockRetryDelay)
}
if !acquired {
return
}
defer registriesMutex.Unlock()
foundAt := -1
@@ -90,8 +123,34 @@ func AddRegistry(registry *DefaultModelRegistry) {
registries = append(registries, registry)
}
// tryLock attempts to acquire the registry's write lock, retrying briefly.
// Returns false if it could not be acquired within the bound.
func (r *DefaultModelRegistry) tryLock() bool {
for i := 0; i < lockRetryAttempts; i++ {
if r.mutex.TryLock() {
return true
}
time.Sleep(lockRetryDelay)
}
return false
}
// tryRLock attempts to acquire the registry's read lock, retrying briefly.
// Returns false if it could not be acquired within the bound.
func (r *DefaultModelRegistry) tryRLock() bool {
for i := 0; i < lockRetryAttempts; i++ {
if r.mutex.TryRLock() {
return true
}
time.Sleep(lockRetryDelay)
}
return false
}
func (r *DefaultModelRegistry) RegisterModel(name string, model interface{}) error {
r.mutex.Lock()
if !r.tryLock() {
return fmt.Errorf("failed to register model %s: registry locked", name)
}
defer r.mutex.Unlock()
if _, exists := r.models[name]; exists {
@@ -137,7 +196,9 @@ func (r *DefaultModelRegistry) RegisterModel(name string, model interface{}) err
}
func (r *DefaultModelRegistry) GetModel(name string) (interface{}, error) {
r.mutex.RLock()
if !r.tryRLock() {
return nil, fmt.Errorf("failed to get model %s: registry locked", name)
}
defer r.mutex.RUnlock()
model, exists := r.models[name]
@@ -149,7 +210,9 @@ func (r *DefaultModelRegistry) GetModel(name string) (interface{}, error) {
}
func (r *DefaultModelRegistry) GetAllModels() map[string]interface{} {
r.mutex.RLock()
if !r.tryRLock() {
return make(map[string]interface{})
}
defer r.mutex.RUnlock()
result := make(map[string]interface{})
@@ -253,14 +316,26 @@ func IterateModels(fn func(name string, model interface{})) {
// GetModels returns a list of all models from all registries
// Models are collected in registry order, with duplicates included
func GetModels() []interface{} {
registriesMutex.RLock()
acquired := false
for i := 0; i < lockRetryAttempts; i++ {
if registriesMutex.TryRLock() {
acquired = true
break
}
time.Sleep(lockRetryDelay)
}
if !acquired {
return nil
}
defer registriesMutex.RUnlock()
var models []interface{}
seen := make(map[string]bool)
for _, registry := range registries {
registry.mutex.RLock()
if !registry.tryRLock() {
continue
}
for name, model := range registry.models {
// Only add the first occurrence of each model name
if !seen[name] {
+11
View File
@@ -17,6 +17,7 @@ package resolvemcp
import (
"net/http"
"runtime/debug"
"github.com/gorilla/mux"
"github.com/uptrace/bun"
@@ -25,6 +26,7 @@ import (
"github.com/bitechdev/ResolveSpec/pkg/common"
"github.com/bitechdev/ResolveSpec/pkg/common/adapters/database"
"github.com/bitechdev/ResolveSpec/pkg/logger"
"github.com/bitechdev/ResolveSpec/pkg/modelregistry"
)
@@ -82,11 +84,20 @@ func SetupMuxRoutes(muxRouter *mux.Router, handler *Handler) {
// - GET {basePath}/sse — SSE connection endpoint
// - POST {basePath}/message — JSON-RPC message endpoint
func SetupBunRouterRoutes(router *bunrouter.Router, handler *Handler) {
defer func() {
if rec := recover(); rec != nil {
logger.Error("panic in resolvemcp.SetupBunRouterRoutes: %v\n%s", rec, debug.Stack())
}
}()
basePath := handler.config.BasePath
h := handler.SSEServer()
router.GET(basePath+"/sse", bunrouter.HTTPHandler(h))
logger.Info("Registered resolvemcp bunrouter route GET %s/sse", basePath)
router.POST(basePath+"/message", bunrouter.HTTPHandler(h))
logger.Info("Registered resolvemcp bunrouter route POST %s/message", basePath)
}
// NewSSEServer returns an http.Handler that serves MCP over SSE.
+70 -2
View File
@@ -3,6 +3,8 @@ package resolvespec
import (
"context"
"fmt"
"sync"
"time"
"github.com/bitechdev/ResolveSpec/pkg/common"
"github.com/bitechdev/ResolveSpec/pkg/logger"
@@ -82,6 +84,7 @@ type HookFunc func(*HookContext) error
// HookRegistry manages all registered hooks
type HookRegistry struct {
hooks map[HookType][]HookFunc
mutex sync.RWMutex
}
// NewHookRegistry creates a new hook registry
@@ -91,8 +94,46 @@ func NewHookRegistry() *HookRegistry {
}
}
// hookLockRetryAttempts/hookLockRetryDelay bound how long the try-lock
// helpers below will spin before giving up, so a contended mutex can
// never hang a caller.
const (
hookLockRetryAttempts = 20
hookLockRetryDelay = 1 * time.Millisecond
)
// tryLock attempts to acquire the write lock, retrying briefly. Returns
// false if it could not be acquired within the bound.
func (r *HookRegistry) tryLock() bool {
for i := 0; i < hookLockRetryAttempts; i++ {
if r.mutex.TryLock() {
return true
}
time.Sleep(hookLockRetryDelay)
}
return false
}
// tryRLock attempts to acquire the read lock, retrying briefly. Returns
// false if it could not be acquired within the bound.
func (r *HookRegistry) tryRLock() bool {
for i := 0; i < hookLockRetryAttempts; i++ {
if r.mutex.TryRLock() {
return true
}
time.Sleep(hookLockRetryDelay)
}
return false
}
// Register adds a new hook for the specified hook type
func (r *HookRegistry) Register(hookType HookType, hook HookFunc) {
if !r.tryLock() {
logger.Error("Failed to register resolvespec hook for %s: registry locked", hookType)
return
}
defer r.mutex.Unlock()
if r.hooks == nil {
r.hooks = make(map[HookType][]HookFunc)
}
@@ -110,8 +151,13 @@ func (r *HookRegistry) RegisterMultiple(hookTypes []HookType, hook HookFunc) {
// Execute runs all hooks for the specified type in order
// If any hook returns an error, execution stops and the error is returned
func (r *HookRegistry) Execute(hookType HookType, ctx *HookContext) error {
hooks, exists := r.hooks[hookType]
if !exists || len(hooks) == 0 {
if !r.tryRLock() {
return fmt.Errorf("hook execution failed: registry locked")
}
hooks := append([]HookFunc(nil), r.hooks[hookType]...)
r.mutex.RUnlock()
if len(hooks) == 0 {
return nil
}
@@ -145,18 +191,35 @@ func (r *HookRegistry) ExecuteBeforeOp(hookType HookType, ctx *HookContext) erro
// Clear removes all hooks for the specified type
func (r *HookRegistry) Clear(hookType HookType) {
if !r.tryLock() {
logger.Error("Failed to clear resolvespec hooks for %s: registry locked", hookType)
return
}
defer r.mutex.Unlock()
delete(r.hooks, hookType)
logger.Info("Cleared all resolvespec hooks for %s", hookType)
}
// ClearAll removes all registered hooks
func (r *HookRegistry) ClearAll() {
if !r.tryLock() {
logger.Error("Failed to clear all resolvespec hooks: registry locked")
return
}
defer r.mutex.Unlock()
r.hooks = make(map[HookType][]HookFunc)
logger.Info("Cleared all resolvespec hooks")
}
// Count returns the number of hooks registered for a specific type
func (r *HookRegistry) Count(hookType HookType) int {
if !r.tryRLock() {
return 0
}
defer r.mutex.RUnlock()
if hooks, exists := r.hooks[hookType]; exists {
return len(hooks)
}
@@ -170,6 +233,11 @@ func (r *HookRegistry) HasHooks(hookType HookType) bool {
// GetAllHookTypes returns all hook types that have registered hooks
func (r *HookRegistry) GetAllHookTypes() []HookType {
if !r.tryRLock() {
return nil
}
defer r.mutex.RUnlock()
types := make([]HookType, 0, len(r.hooks))
for hookType := range r.hooks {
types = append(types, hookType)
+17
View File
@@ -2,6 +2,7 @@ package resolvespec
import (
"net/http"
"runtime/debug"
"strings"
"github.com/gorilla/mux"
@@ -12,6 +13,7 @@ import (
"github.com/bitechdev/ResolveSpec/pkg/common"
"github.com/bitechdev/ResolveSpec/pkg/common/adapters/database"
"github.com/bitechdev/ResolveSpec/pkg/common/adapters/router"
"github.com/bitechdev/ResolveSpec/pkg/logger"
"github.com/bitechdev/ResolveSpec/pkg/modelregistry"
)
@@ -244,6 +246,11 @@ func wrapBunRouterHandler(handler bunrouter.HandlerFunc, authMiddleware Middlewa
// Accepts bunrouter.Router or bunrouter.Group
// authMiddleware is optional - if provided, routes will be protected with the middleware
func SetupBunRouterRoutes(r BunRouterHandler, handler *Handler, authMiddleware MiddlewareFunc) {
defer func() {
if rec := recover(); rec != nil {
logger.Error("panic in resolvespec.SetupBunRouterRoutes: %v\n%s", rec, debug.Stack())
}
}()
// CORS config
corsConfig := common.DefaultCORSConfig()
@@ -269,6 +276,13 @@ func SetupBunRouterRoutes(r BunRouterHandler, handler *Handler, authMiddleware M
// Loop through each registered model and create explicit routes
for fullName := range allModels {
func() {
defer func() {
if rec := recover(); rec != nil {
logger.Error("panic registering resolvespec routes for model %s: %v\n%s", fullName, rec, debug.Stack())
}
}()
// Parse the full name (e.g., "public.users" or just "users")
schema, entity := parseModelName(fullName)
@@ -375,6 +389,9 @@ func SetupBunRouterRoutes(r BunRouterHandler, handler *Handler, authMiddleware M
handler.HandleGet(respAdapter, reqAdapter, params)
return nil
})
logger.Info("Registered resolvespec bunrouter routes for model %s at %s", fullName, entityPath)
}()
}
}
+70 -2
View File
@@ -3,6 +3,8 @@ package restheadspec
import (
"context"
"fmt"
"sync"
"time"
"github.com/bitechdev/ResolveSpec/pkg/common"
"github.com/bitechdev/ResolveSpec/pkg/logger"
@@ -89,6 +91,7 @@ type HookFunc func(*HookContext) error
// HookRegistry manages all registered hooks
type HookRegistry struct {
hooks map[HookType][]HookFunc
mutex sync.RWMutex
}
// NewHookRegistry creates a new hook registry
@@ -98,8 +101,46 @@ func NewHookRegistry() *HookRegistry {
}
}
// hookLockRetryAttempts/hookLockRetryDelay bound how long the try-lock
// helpers below will spin before giving up, so a contended mutex can
// never hang a caller.
const (
hookLockRetryAttempts = 20
hookLockRetryDelay = 1 * time.Millisecond
)
// tryLock attempts to acquire the write lock, retrying briefly. Returns
// false if it could not be acquired within the bound.
func (r *HookRegistry) tryLock() bool {
for i := 0; i < hookLockRetryAttempts; i++ {
if r.mutex.TryLock() {
return true
}
time.Sleep(hookLockRetryDelay)
}
return false
}
// tryRLock attempts to acquire the read lock, retrying briefly. Returns
// false if it could not be acquired within the bound.
func (r *HookRegistry) tryRLock() bool {
for i := 0; i < hookLockRetryAttempts; i++ {
if r.mutex.TryRLock() {
return true
}
time.Sleep(hookLockRetryDelay)
}
return false
}
// Register adds a new hook for the specified hook type
func (r *HookRegistry) Register(hookType HookType, hook HookFunc) {
if !r.tryLock() {
logger.Error("Failed to register hook for %s: registry locked", hookType)
return
}
defer r.mutex.Unlock()
if r.hooks == nil {
r.hooks = make(map[HookType][]HookFunc)
}
@@ -117,8 +158,13 @@ func (r *HookRegistry) RegisterMultiple(hookTypes []HookType, hook HookFunc) {
// Execute runs all hooks for the specified type in order
// If any hook returns an error, execution stops and the error is returned
func (r *HookRegistry) Execute(hookType HookType, ctx *HookContext) error {
hooks, exists := r.hooks[hookType]
if !exists || len(hooks) == 0 {
if !r.tryRLock() {
return fmt.Errorf("hook execution failed: registry locked")
}
hooks := append([]HookFunc(nil), r.hooks[hookType]...)
r.mutex.RUnlock()
if len(hooks) == 0 {
// logger.Debug("No hooks registered for %s", hookType)
return nil
}
@@ -154,18 +200,35 @@ func (r *HookRegistry) ExecuteBeforeOp(hookType HookType, ctx *HookContext) erro
// Clear removes all hooks for the specified type
func (r *HookRegistry) Clear(hookType HookType) {
if !r.tryLock() {
logger.Error("Failed to clear hooks for %s: registry locked", hookType)
return
}
defer r.mutex.Unlock()
delete(r.hooks, hookType)
logger.Info("Cleared all hooks for %s", hookType)
}
// ClearAll removes all registered hooks
func (r *HookRegistry) ClearAll() {
if !r.tryLock() {
logger.Error("Failed to clear all hooks: registry locked")
return
}
defer r.mutex.Unlock()
r.hooks = make(map[HookType][]HookFunc)
logger.Info("Cleared all hooks")
}
// Count returns the number of hooks registered for a specific type
func (r *HookRegistry) Count(hookType HookType) int {
if !r.tryRLock() {
return 0
}
defer r.mutex.RUnlock()
if hooks, exists := r.hooks[hookType]; exists {
return len(hooks)
}
@@ -179,6 +242,11 @@ func (r *HookRegistry) HasHooks(hookType HookType) bool {
// GetAllHookTypes returns all hook types that have registered hooks
func (r *HookRegistry) GetAllHookTypes() []HookType {
if !r.tryRLock() {
return nil
}
defer r.mutex.RUnlock()
types := make([]HookType, 0, len(r.hooks))
for hookType := range r.hooks {
types = append(types, hookType)
+16
View File
@@ -55,6 +55,7 @@ package restheadspec
import (
"net/http"
"runtime/debug"
"strings"
"github.com/gorilla/mux"
@@ -308,6 +309,11 @@ func wrapBunRouterHandler(handler bunrouter.HandlerFunc, authMiddleware Middlewa
// Accepts bunrouter.Router or bunrouter.Group
// authMiddleware is optional - if provided, routes will be protected with the middleware
func SetupBunRouterRoutes(r BunRouterHandler, handler *Handler, authMiddleware MiddlewareFunc) {
defer func() {
if rec := recover(); rec != nil {
logger.Error("panic in restheadspec.SetupBunRouterRoutes: %v\n%s", rec, debug.Stack())
}
}()
// CORS config
corsConfig := common.DefaultCORSConfig()
@@ -333,6 +339,13 @@ func SetupBunRouterRoutes(r BunRouterHandler, handler *Handler, authMiddleware M
// Loop through each registered model and create explicit routes
for fullName := range allModels {
func() {
defer func() {
if rec := recover(); rec != nil {
logger.Error("panic registering restheadspec routes for model %s: %v\n%s", fullName, rec, debug.Stack())
}
}()
// Parse the full name (e.g., "public.users" or just "users")
schema, entity := parseModelName(fullName)
@@ -498,6 +511,9 @@ func SetupBunRouterRoutes(r BunRouterHandler, handler *Handler, authMiddleware M
handler.HandleGet(respAdapter, reqAdapter, params)
return nil
})
logger.Info("Registered restheadspec bunrouter routes for model %s at %s", fullName, entityPath)
}()
}
}