miyako

README

🌸 miyako

The Metropolitan Concurrency & Tactical Response Engine for Go

"Every empire needs a capital. Where raw goroutines run like chaotic crowds, Miyako provides the absolute architecture, administrative order, and high-speed transit of a pristine metropolis."

🇺🇸 English • 🇷🇺 Русский

Why Miyako?

When building microservices, real-time event loops, or distributed worker pools in Go, managing concurrency can feel like navigating an unpredictable, sprawling wilderness. Without strict coordination, goroutine leaks, database race conditions, and resource deadlocks threaten to destabilize your entire system.

Named after the historic capital (miyako / 都) and built on the principles of metropolitan order and unwavering discipline, miyako is a high-performance Go toolkit. It acts as the administrative backbone of your application — providing structured service orchestrators (lifecycle), high-speed concurrent pipelines (yumi), and secure state guards (keylock) to govern millions of concurrent operations with absolute urban-grade precision.

go get github.com/lemon4ksan/miyako

🎯 When to Use Miyako vs. Standard Primitives

miyako is engineered for complex, high-risk coordination scenarios where state corruption or runtime deadlocks could cause system-wide crashes.

• Choose standard sync / channels for: Simple pipelines, basic worker pools, local mutex blocks, and standard short-lived async operations.
• Choose miyako for: Topologically sorted service boot sequences, correlation-ID job tracking, non-blocking type-safe event backbones, striped key-based locking with automatic cleanup, and quick-draw request deduplication. It is your combat gear for hostile concurrent environments.

⚡ The Contrast: Raw Go vs. miyako

Every miyako package replaces a pile of boilerplate, manual locking, and silent runtime failures with a concise, generics-first, thread-safe API. Here is what you are writing today versus what you could be writing.

Job Tracking

Raw Go (Manual State Tracking)	Using miyako
type Job struct { Done chan struct{} Err error } mu.Lock() jobs[id] = job mu.Unlock() go func() { select { case <-ctx.Done(): case <-time.After(timeout): } }() // manual error propagation & wait loop	mgr := jobs.NewManager[string, Result](capacity) err := mgr.Add(jobID, callback, jobs.WithTimeout[Result](30*time.Second), jobs.WithContext[Result](ctx), ) res, err := mgr.WaitFor(ctx, jobID)

Request Deduplication

Raw Go (Manual Dedup)

Using miyako

// 🔴 Panic in worker kills ALL waiters // 🔴 No context cancellation // 🔴 Manual map + mutex + channel wiring var ( mu sync.Mutex inflight = map[string]*call{} ) type call struct { wg sync.WaitGroup val *User err error } func fetchUser(key string) (*User, error) { mu.Lock() if c, ok := inflight[key]; ok { mu.Unlock() c.wg.Wait() return c.val, c.err } c := &call{wg: sync.WaitGroup{1}} inflight[key] = c mu.Unlock() c.wg.Add(1) c.val, c.err = db.Fetch(key) // panic = all waiters die c.wg.Done() mu.Lock() delete(inflight, key) mu.Unlock() return c.val, c.err }

// ✅ Panic isolated to initiator only // ✅ Context cancellation on all waiters // ✅ Zero-value ready, no setup group := &batto.Group[string, *User]{} user, err := group.Do(ctx, "user-123", func(ctx context.Context) (*User, error) { return db.FetchUser(ctx, 123) }, )

Per-Key Locking

Raw Go (Memory-Leaky Lock Map)

Using miyako

// 🔴 Mutexes never cleaned up → memory leak // 🔴 No TryLock, no ForceUnlock // 🔴 Manual bookkeeping for every key var ( mu sync.Mutex locks = map[string]*sync.Mutex{} ) func getLock(key string) *sync.Mutex { mu.Lock() defer mu.Unlock() if locks[key] == nil { locks[key] = &sync.Mutex{} } return locks[key] } func processOrder(orderID string) { getLock(orderID).Lock() defer getLock(orderID).Unlock() // ... work ... // 🔑 lock entry stays in map forever }

// ✅ Auto-cleanup via refcount when key is released // ✅ TryLock, ForceUnlock, Keys() built-in // ✅ Generic key type: string, int, UUID, etc. lock := keylock.New[string]() func processOrder(orderID string) { lock.Lock(orderID) defer lock.Unlock(orderID) // ... work ... // 🔑 entry auto-deleted when refcount hits 0 }

Lazy Initialization

Raw Go (sync.Once)	Using miyako
// 🔴 No reset - once broken, broken forever // 🔴 Must store value + once separately var ( dbOnce sync.Once db *sql.DB ) func getDB() *sql.DB { dbOnce.Do(func() { var err error db, err = sql.Open("pg", dsn) if err != nil { // 🔴 dbOnce already marked done // 🔴 subsequent calls return nil, nil } }) return db } // getDB() after error = nil forever	// ✅ Reset re-runs initialization on next Get() // ✅ Thread-safe, generic, zero-value ready db := lazy.New(func() *sql.DB { conn, _ := sql.Open("pg", dsn) return conn }) func getDB() *sql.DB { return db.Get() } // After error recovery: db.Reset() // next Get() retries initialization

Bulk Parallel Processing

Raw Go (Manual Fan-Out)

Using miyako

// 🔴 Order not preserved // 🔴 No rate limiting // 🔴 Manual WaitGroup + error collection results := make([]Result, len(items)) var wg sync.WaitGroup sem := make(chan struct{}, 10) for _, item := range items { wg.Add(1) sem <- struct{}{} go func(it Item) { defer wg.Done() defer func() { <-sem }() r, err := process(ctx, it) // 🔴 race on results slice if no mutex results[idx] = r }(item) } wg.Wait() // 🔴 results order != items order

// ✅ Order preserved, rate-limited, fail-fast option // ✅ One-liner for slice processing results, err := yumi.Map(ctx, yumi.PipelineConfig{ Workers: 10, RPS: 100, FailFast: true, }, items, func(ctx context.Context, it Item) (Result, error) { return process(ctx, it) }) // results order == items order

Concurrency Limiting

Raw Go (Static Semaphore)	Using miyako
// 🔴 Limit is fixed at creation // 🔴 No dynamic resize // 🔴 Zombie goroutines on ctx cancel sem := make(chan struct{}, 10) // Acquire sem <- struct{}{} // Release <-sem // 🔴 Changing limit requires recreating channel // 🔴 ctx cancel doesn't unblock waiting goroutines	// ✅ Dynamic resize without restart // ✅ ctx cancellation unblocks waiters instantly // ✅ Clean API sem := semaphore.New(10) if err := sem.Acquire(ctx); err != nil { return err // ctx cancelled → clean exit } defer sem.Release() // Later: scale up/down at runtime sem.Resize(20)

Behavior Orchestration

Raw Go (Manual Goroutine Management)

Using miyako

// 🔴 Manual goroutine tracking // 🔴 No fail-fast, no graceful shutdown // 🔴 Error in one goroutine = silent leak var wg sync.WaitGroup ctx, cancel := context.WithCancel(ctx) wg.Add(2) go func() { defer wg.Done() for { select { case <-ctx.Done(): return default: if err := ticker(); err != nil { log.Println(err) // 🔴 other goroutines keep running } } } }() go func() { defer wg.Done() for { select { case <-ctx.Done(): return default: if err := watcher(); err != nil { log.Println(err) } } } }() cancel() wg.Wait()

// ✅ Managed lifecycle with fail-fast // ✅ Graceful shutdown, all goroutines tracked // ✅ Logger integration, duplicate detection runner := lifecycle.NewBehaviorRunner( lifecycle.WithLogger(myLogger), lifecycle.WithFailFast(), ) runner.Register(&tickerBehavior{}) runner.Register(&watcherBehavior{}) ctx, cancel := context.WithCancel(ctx) defer cancel() runner.Start(ctx) // ... later runner.Stop() // all behaviors stopped cleanly

What Every Package Replaces

Package	You write today (Raw Go)	miyako equivalent
batto	sync.Mutex + map[string]*call + sync.WaitGroup + panic recovery	batto.Group[K, V]{}.Do(ctx, key, fn)
bus	Channels per type + reflect switch + manual fan-out	bus.New() → Subscribe / Publish
jobs	Goroutine + chan + sync.Mutex + time.After + manual cleanup	jobs.NewManager[K, T](n) → Add / WaitFor
lifecycle	Hardcoded init order + manual rollback on failure, or manual wg runner	lifecycle.NewOrchestrator() → Register / StartAll, and lifecycle.NewBehaviorRunner()
scheduler	time.Ticker + sort.Slice + manual wake-up loop	scheduler.New() → Schedule / Start
yumi	sync.WaitGroup + buffered channel + sync.Mutex for results	yumi.Map(ctx, cfg, items, fn)
semaphore	make(chan struct{}, N) - static, no cancel	semaphore.New(n) → Acquire(ctx) / Resize
keylock	map[string]*sync.Mutex - no cleanup, no TryLock	keylock.New[K]() → Lock(key)
lazy	sync.Once + separate var - no reset	lazy.New(fn) → Get() / Reset()
spinlock	sync.Mutex - heavier for short critical sections	spinlock.SpinLock{} → Lock() / Unlock()
generic	Duplicated map/filter/retry per package	generic.Map, generic.Retry, generic.Future
limiter	sync.Mutex + golang.org/x/time/rate - no key-based auto-cleanup, static algorithm	limiter.NewAdaptiveLimiter(N) / limiter.NewKeyedLimiter[K](...)
breaker	sony/gobreaker - reflection/empty interfaces, no generics	breaker.New[T](cfg) → Do
pool	channel/WaitGroup worker pool - static, no idle scale-down / panic safety	pool.NewPool[T](cfg) → Submit

📊 Feature Matrix

This matrix shows where miyako focuses its design compared to Go's default primitives and generic wrappers:

Feature / Capability	Go sync (StdLib)	Go x/sync (Experimental)	miyako
Generics-first Design	✗ (Manual)	✗ (Interface-based)	✓ (Type-safe [T])
Topological Startup & Shutdown	✗	✗	✓ (lifecycle.Orchestrator)
Correlation-ID Job Tracking	✗	✗	✓ (jobs.Manager)
Cancellable Resizable Semaphore	✗	⚠️ (Static only)	✓ (sync/semaphore.Semaphore)
Request Deduplication	✗	⚠️ (basic SingleFlight)	✓ (batto.Group / Quick-Draw)
Key-Based Striped Mutex	✗	✗	✓ (sync/keylock.KeyMutex)
Non-Blocking Type-Based Event Bus	✗	✗	✓ (bus.Bus / Type-Safe)
Reset-Aware Lazy Initializer	⚠️ (sync.Once)	✗	✓ (sync/lazy.Lazy)
Ultra-Fast Spinlock Waiting	✗	✗	✓ (sync/spinlock.SpinLock)
Batching Request DataLoader	✗	✗	✓ (generic.DataLoader)
Strict Generic State Machine	✗	✗	✓ (kata.FSM)
Concurrent Behavior Runner	✗	✗	✓ (lifecycle.BehaviorRunner)
Vegas Congestion Limiter	✗	✗	✓ (sync/limiter.AdaptiveLimiter)
Keyed Limiter with Auto-Cleanup	✗	✗	✓ (sync/limiter.KeyedLimiter)
Generics-first Circuit Breaker	✗	✗	✓ (sync/breaker.CircuitBreaker)
Auto-Scaling Dynamic Worker Pool	✗	✗	✓ (pool.Pool)

🔍 Under the Hood: Architecture & Algorithms

To remove the fear of "magic" and explain how the toolkit operates mathematically, here is a detailed breakdown of the internal design and algorithms behind each miyako subpackage:

1. batto (Single-Flight Group)

• Algorithm: Coordinates concurrent executions for identical parameterized keys.
• Mechanism: When a key-parameterized function is executed, it first checks an internal map under a sync.Mutex lock. If a call is already in progress, the caller allocates a channel, registers it as a waiter, and blocks. When the active worker completes, it distributes the result or propagates panic to the initiating waiter, notifying secondary waiters with ErrWorkerPanicked. Waiter channels are swept when reference count drops to 0.

2. bus (Event Bus)

• Algorithm: A type-safe event bus with non-blocking event publishing.
• Mechanism: It keeps a registry of subscribers mapped by the reflect.Type of events. Subscriptions have buffered Go channels. On Publish, the bus retrieves subscribers matching the event type and sends the event to their channels in a non-blocking/buffered manner. Unsubscription deletes elements from the slice of subscribers associated with that type under lock protection.

3. generic (Generic Utilities)

• Algorithm: A collection of functional slice mapping/filtering, retry loops with backoff, and standard promise patterns (Future) using Go standard channels.
• Mechanism: Functions like ParallelMap limit active goroutines with buffered channels functioning as semaphores. Backoff implements the AWS Full Jitter algorithm for exponential backoff, calculating retry windows with randomized deviation to prevent thundering herd problems.

4. jobs (Correlation-ID Job Manager)

• Algorithm: Correlation-ID based job tracking with asynchronous callback execution.
• Mechanism: Registers expected jobs in a thread-safe map before calling an external asynchronous system. When Resolve is called with the correlation ID, the manager marks the job completed, writes to the job's channel to unblock WaitFor, and runs its callback asynchronously according to a configured strategy. Auto-reclaims resources via a sync.Pool for job entries.

5. kata (State Machine)

• Algorithm: Strictly typed finite state machine (FSM).
• Mechanism: It represents a state transition matrix using a map of (State, Event) -> State transitions. Transitions are guarded by a read-write lock. OnBefore and OnAfter hooks can abort/rollback the transition atomically if they return an error.

6. lifecycle (Service Orchestration & Background Loops)

• Algorithm:
  • Orchestrator: Builds a Directed Acyclic Graph (DAG) of dependencies and performs a Depth-First Search (DFS) topological sort. Services are initialized and started in this sorted order. On startup failure, all running services are stopped in reverse order.
  • BehaviorRunner: Registers background loops and runs them in separate goroutines, managing graceful shutdown and optional fail-fast cancellation.

7. log (Asynchronous Logger)

• Algorithm: Formats logs on the calling thread into a buffer retrieved from a sync.Pool to avoid allocations, then writes the buffer to a buffered Go channel.
• Mechanism: A background goroutine consumes from this channel and writes to the configured io.Writer. Drops logs if the buffer queue size limits are breached to avoid blocking.

8. pool (Auto-Scaling Worker Pool)

• Algorithm: Submits tasks as a work function packaged with a local Future to a buffered task channel.
• Mechanism: If all current workers are busy, it scales up worker goroutines up to MaxWorkers. Idle workers beyond MinWorkers monitor the task channel with a time.Timer and exit if they receive no tasks within IdleTimeout. Task panics are caught inside workers and stored in the task's Future.

9. scheduler (Job Scheduler)

• Algorithm: Maintains scheduled jobs in a list sorted by execution time.
• Mechanism: An internal timer sleep loop wakes up whenever the next task is due. Execution is dispatched to concurrent goroutines.

10. sync (Synchronization Primitives)

• breaker: Tracks success/failure ratio of requests in a sliding time window (pruned on each invocation). If the failure ratio over MinRequests exceeds the threshold, the state machine transitions to StateOpen and records openTime, failing subsequent requests fast. After a Cooldown period, it enters StateHalfOpen and allows a single trial request to proceed.
• keylock: Holds a map of mutexes with reference counters. When Lock is requested, it increments the reference counter for that key and acquires the lock. On Unlock, it releases the lock, decrements the counter, and deletes the key-mutex mapping when counter reaches 0 under lock protection.
• lazy: Wraps sync.Once and a value. A Reset() call resets the state by instantiating a new internal sync.Once and clearing the value, permitting re-initialization on subsequent calls.
• limiter:
  • AdaptiveLimiter: Vegas congestion control algorithm. It monitors RTT (Round Trip Time). If RTT increases beyond the queue limit, it scales down the concurrency limit; if RTT is low, it scales up concurrency.
  • KeyedLimiter: Holds a map of rate limiters with clean-up TTLs. A background sweeper sweeps limiters that haven't been accessed within TTL.
• semaphore: Managed via slots and channels. Modifying limits resizes the internal token queue dynamically, adding or draining tokens, while wait queue blocks on Go channels that are fully responsive to context cancellation.
• spinlock: Performs busy-waiting loop using runtime.Gosched() for a small count of loops before backing off, bypassing system-level thread switching overhead for extremely short critical sections.

11. yumi (Concurrent Mapping/Pipeline)

• Algorithm: Spawns a limited pool of worker goroutines.
• Mechanism: It processes slice elements concurrently, writing results to an indexed slice so that order is strictly preserved. Includes a token-bucket rate limiter for RPS enforcement, and aborts processing early if FailFast configuration is activated.

🍳 The Concurrency Kata: Tactical Recipes

Here is how you solve common, frustrating concurrency and orchestration challenges using miyako.

1. Battojutsu Request Deduplication (batto)

• The Problem: Multiple incoming API requests query the same database entry concurrently, spawning expensive SQL calls. If the SQL query panics, standard singleflight will panic all waiting threads, or leak them.
• The Solution: Named after Battojutsu (拔刀术 / the art of quick-drawing a katana), the batto package cuts off duplicate concurrent calls. If the worker panics, the panic is safely isolated and propagated only to the initiating thread, while secondary waiters receive a clean ErrWorkerPanicked.

group := &batto.Group[string, *User]{}

user, err := group.Do(ctx, "user-123", func(workerCtx context.Context) (*User, error) {
    // Spawns exactly once for concurrent requests to "user-123"
    return db.FetchUser(workerCtx, 123)
})

2. Topologically Sorted Service Bootstrapping (lifecycle)

• The Problem: Your microservice needs Database started first, then RedisCache (which depends on database), and finally the WebServer (which depends on both). During shutdown, they must stop in the exact reverse order.
• The Solution: lifecycle uses a Depth-First Search (DFS) algorithm to sort and initialize your services, resolving dependencies and rolling back automatically on failure.

orchestrator := lifecycle.NewOrchestrator()

// Register services. Dependents declare their dependencies.
orchestrator.Register(NewDatabaseService())
orchestrator.Register(NewRedisCacheService()) // Dependencies() -> []string{"db"}
orchestrator.Register(NewWebServerService())  // Dependencies() -> []string{"db", "redis"}

// Sorts topologically and initializes all services
if err := orchestrator.InitAll(ctx); err != nil {
    log.Fatalf("Init failed: %v", err)
}

// Starts all services. On any failure, successfully started services rollback in reverse.
if err := orchestrator.StartAll(ctx); err != nil {
    log.Fatalf("Start failed: %v", err)
}

// Gracefully shuts down in reverse topological order: WebServer -> RedisCache -> Database
defer orchestrator.StopAll(context.Background())

3. Dynamic Concurrency Gating (Cancellable Resizable Semaphore)

• The Problem: Your worker pool needs to limit concurrent calls to an upstream API. The API capacity changes dynamically during runtime, and waiting workers must unblock immediately if their context is canceled.
• The Solution: sync/semaphore manages dynamic limits and includes bulletproof context cancellation, preventing zombie channels from leaking memory.

// Create a semaphore with an initial limit of 10 concurrent requests
sem := semaphore.New(10)

go func() {
    // Adapt limit dynamically based on upstream API health signals
    time.Sleep(1 * time.Minute)
    sem.Resize(5) // Scale down capacity to 5 slots
}()

// Acquire slot respects context cancellation without leaving zombie channels
if err := sem.Acquire(ctx); err != nil {
    return err // Context cancelled, worker exits cleanly
}
defer sem.Release()

api.Call()

4. Striped Locking with Auto-Teardown (sync/keylock)

• The Problem: You want to serialize operations per user ID. Storing a standard sync.Mutex per user in a global map will cause a memory leak as users connect and disconnect.
• The Solution: keylock manages dynamic mutexes and automatically cleans them up from memory once the reference count of waiting goroutines drops to zero.

lock := keylock.New[string]()

func ProcessUserRecord(userID string) {
    lock.Lock(userID)
    defer lock.Unlock(userID) // KeyMutex automatically deletes key from map when count == 0
    
    // Serialized user record modifications
}

5. Automated Rate-Limited Batching (generic.DataLoader)

• The Problem: Multiple goroutines query individual item metrics concurrently, causing your API client to get rate-limited.
• The Solution: Named after Yumi (弓 / the Japanese bow), generic.DataLoader collects individual queries over a 5ms window, batches them into a single bulk query, and distributes results back to each goroutine.

loader := yumi.NewDataLoader[string, *Price](5*time.Millisecond, func(ctx context.Context, keys []string) (map[string]*Price, error) {
    return pricedbClient.GetItemsBulk(ctx, keys) // Executed exactly once!
})

// Spawning this concurrently in 10 goroutines will trigger only ONE API call!
price, err := loader.Load(ctx, "item_sku")

6. Strict Generic State Machine (kata)

• The Problem: You need to model a lifecycle (e.g. order processing, connection states, bot flow) where invalid transitions must be caught at compile time, and all state changes must be thread-safe with transactional rollback support.
• The Solution: kata provides a strictly typed FSM parameterized over comparable State and Event generics. It supports before/after hooks with rollback, concurrent-safe transitions, and automatic Graphviz DOT export.

type State int
const (
    Idle State = iota
    Running
    Stopped
)

type Event int
const (
    Start Event = iota
    Stop
)

fsm := kata.NewFSM[State, Event](Idle)

fsm.AddRules(
    kata.TransitionRule[State, Event]{From: Idle, Event: Start, To: Running},
    kata.TransitionRule[State, Event]{From: Running, Event: Stop, To: Stopped},
)

// Before-hook: abort transition if preconditions fail
fsm.OnBefore(Start, func(ctx context.Context, from State, event Event, to State) error {
    if !healthCheckOK(ctx) {
        return errors.New("upstream unhealthy, blocking start")
    }
    return nil
})

// Thread-safe from any goroutine
err := fsm.Transition(context.Background(), Start)

// Export visual diagram
fmt.Println(fsm.ToDOT())

7. Concurrent Behavior Orchestration (lifecycle.BehaviorRunner)

• The Problem: You need to run multiple independent background tasks (ticker, watcher, health-check) in parallel, with coordinated shutdown and optional fail-fast - but manual sync.WaitGroup + context.WithCancel wiring is error-prone and untestable.
• The Solution: lifecycle provides a BehaviorRunner that manages the lifecycle of registered Behavior instances. Each runs in its own goroutine with automatic tracking, graceful shutdown, and optional fail-fast mode.

type tickerBehavior struct {
    name     string
    interval time.Duration
}

func (t *tickerBehavior) Name() string { return t.name }

func (t *tickerBehavior) Run(ctx context.Context) error {
    ticker := time.NewTicker(t.interval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return nil
        case <-ticker.C:
            fmt.Printf("Tick from %s\n", t.name)
        }
    }
}

runner := lifecycle.NewBehaviorRunner(
    lifecycle.WithFailFast(),
)

runner.Register(&tickerBehavior{name: "fast", interval: time.Second})
runner.Register(&tickerBehavior{name: "slow", interval: 5 * time.Second})

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

runner.Start(ctx)
// ... later
runner.Stop() // all behaviors stopped cleanly

8. Dynamic Vegas-Style Concurrency Limiting (sync/limiter.AdaptiveLimiter)

• The Problem: Setting a static concurrency limit (like a semaphore) on a client calling an upstream API leads to either underutilization (limit too low during healthy periods) or overloading the upstream (limit too high during spikes).
• The Solution: sync/limiter.AdaptiveLimiter dynamically adjusts the concurrent slot limit by measuring response times (RTT). It automatically shrinks the limit when the upstream gets congested (RTT rises) and grows it when the upstream is healthy.

// Start with an initial concurrency limit of 10.0
lim := limiter.NewAdaptiveLimiter(10.0)

func HandleRequest(ctx context.Context) error {
    if err := lim.Acquire(ctx); err != nil {
        return err // Context timeout or cancellation
    }

    start := time.Now()
    err := callUpstreamAPI(ctx)
    rtt := time.Since(start)

    // Release slot and update RTT metrics to dynamically recalculate the limit
    lim.Release(rtt)

    return err
}

9. Keyed Rate Limiter with Auto-Teardown (sync/limiter.KeyedLimiter)

• The Problem: You need to rate-limit users by their API key or IP address. Storing a standard rate.Limiter per key in a map will leak memory over time as unique keys connect and never return.
• The Solution: limiter.KeyedLimiter dynamically allocates rate limiters for active keys and automatically sweeps them from memory after a configured TTL of inactivity.

// Rate limit: 5 requests/sec, burst size 10, TTL 5 minutes
kl := limiter.NewKeyedLimiter[string](rate.Limit(5), 10, 5*time.Minute)
defer kl.Close()

func HandleUserRequest(ctx context.Context, userID string) error {
    // Dynamically retrieves/creates limiter for userID and resets its inactivity TTL
    if err := kl.Wait(ctx, userID); err != nil {
        return err // Limit exceeded or context cancelled
    }

    return processRequest()
}

10. Generics-First Circuit Breaker (sync/breaker.CircuitBreaker)

• The Problem: Downstream microservice outages can cascade, exhausting caller threads. Standard breakers rely on legacy reflection or interface{} casting, which is verbose and slow.
• The Solution: breaker.CircuitBreaker wraps execution with strict compile-time type safety, failing fast when failure rates exceed thresholds, and performing safe single-request trials in Half-Open state.

cb := breaker.New[User](breaker.Config{
    FailureThreshold: 0.5,              // Trip when 50% of requests fail
    Cooldown:         10 * time.Second, // Wait 10s before testing Half-Open
    MinRequests:      5,                // Gather at least 5 requests before tripping
})

user, err := cb.Do(ctx, func(ctx context.Context) (User, error) {
    return userClient.Fetch(ctx)
})
if err != nil {
    if errors.Is(err, breaker.ErrCircuitOpen) {
        // Fast-path recovery or fallback
        return getCachedUser(), nil
    }

    return User{}, err
}

11. Dynamic Worker Pool with Auto-Scaling (pool.Pool)

• The Problem: Static worker pools waste system resources by keeping idle goroutines alive during quiet hours, or bottleneck the pipeline when tasks spike.
• The Solution: pool.Pool scales worker goroutines dynamically between MinWorkers and MaxWorkers. It kills idle workers after an IdleTimeout and isolates task panics to protect the pool structure.

p := pool.New[int](pool.Config{
    MinWorkers:  2,
    MaxWorkers:  20,
    IdleTimeout: 10 * time.Second,
    QueueLimit:  100,
})
defer p.Close()

// Submit task and receive a Future
future, err := p.Submit(ctx, func(ctx context.Context) (int, error) {
    return performHeavyTask(ctx)
})
if err != nil {
    if errors.Is(err, pool.ErrQueueFull) {
        // Handle overflow (fail fast)
    }

    return err
}

// Block and retrieve value when ready
result, err := future.Get(ctx)

🔬 The Contrast: Raw FSM vs. kata

A finite state machine without generics forces you into interface{} or string states, manual locking, and scattered transition logic. Here is what that looks like:

Raw Go FSM (Boilerplate & Unsafe)

Using kata (Generics & Thread-Safe)

// 🔴 No compile-time safety: states are strings // 🔴 Manual mutex on every access // 🔴 No hooks, no rollback, no visualization type RawFSM struct { mu sync.RWMutex current string rules map[string]map[string]string } func (f *RawFSM) Transition(event string) error { f.mu.Lock() defer f.mu.Unlock() events, ok := f.rules[f.current] if !ok { return fmt.Errorf("no rules for %s", f.current) } to, ok := events[event] if !ok { return fmt.Errorf("invalid: %s + %s", f.current, event) } f.current = to return nil } // Usage: easy to typo, no compiler help fsm := &RawFSM{ current: "idle", rules: map[string]map[string]string{ "idle": {"start": "running"}, }, } fsm.Transition("statr") // typo - compiles fine, runtime error

// ✅ Compile-time safe: wrong state = build error // ✅ Thread-safe by design, no manual locks // ✅ Before/after hooks, rollback, DOT export fsm := kata.NewFSM[State, Event](Idle) fsm.AddRules( kata.TransitionRule[State, Event]{ From: Idle, Event: Start, To: Running, }, ) fsm.OnBefore(Start, func(ctx context.Context, from State, event Event, to State) error { return db.BeginTx(ctx) // rollback on error }) // Usage: typos caught at compile time fsm.Transition(Start) // OK fsm.Transition(Statr) // COMPILE ERROR

What you get with kata that raw implementations lack:

Concern	Raw Go FSM	kata FSM
Type safety	string / interface{} - typos are silent	Generic [State, Event] - wrong types won't compile
Thread safety	Manual sync.Mutex on every method	Built-in sync.RWMutex, lock-free reads
Transition hooks	Scattered if checks before/after	OnBefore / OnAfter with rollback support
Transactional rollback	Manual flag + restore on error	Before-hook error aborts atomically
Validation	Runtime panic or silent miss	Validate() + compile-time guarantees
Visualization	Draw diagrams by hand	ToDOT() - one line, render with Graphviz
Test setup	Rewrite transition logic in test helpers	ForceSet() - direct state injection

⚖️ Legal & License

This project is licensed under the BSD 3-Clause License. See LICENSE for full details.

_{Keep a cold head, protect the capital. Discipline of Section 6.}