README
¶
Coalesce 🚀
coalesce is a high-performance, type-safe request coalescing and cache stampede (thundering herd) protection library for Go (1.21+).
It elegantly collapses thousands of concurrent, redundant, or heavy operations (such as database queries, microservice calls, or resource initializations) into a single execution, while providing surgical-grade resource management through dynamic orphan task auto-cancellation.
🌟 Core Features
- 🛡️ Thundering Herd Shield: Collapses mass concurrent flights into exactly one execution.
- ⚡ Type-Safe Generics: Built from the ground up utilizing Go 1.21+ generics for compile-time safety—no more
interface{}casting. - 🛑 Smart Auto-Cancellation (Orphan Teardown): If all waiting clients timeout or abort,
coalesceautomatically cuts off the background worker via context propagation, eliminating ghost tasks from draining your system resources. - 🔄 Resilient Failure Isolation: Failed background initializations are instantly evicted from tracking states, allowing immediate retries without polluting upper cache tiers.
- 🔀 Atomic Cache Handover: Built-in hooks natively synchronize results with your infrastructure storage (e.g., Redis, LRU) safely with zero-copy friction.
📦 Installation
go get -u github.com/gophini/coalesce
🧩 Key Components
| Component | Architecture Role | Best Used For |
|---|---|---|
coalesce.Task[T] |
Single-resource lazy coordinator | Global singletons, heavy application configs, DB connection pool setups. |
coalesce.Group[K, V] |
Keyspaced dynamic synchronization center | Dynamic cache warming, HTTP/gRPC hot-key protection, dynamic metadata fetching. |
💡 Quick Start
1. Guarding a Keyspaced Cache Placement (coalesce.Group)
This is the standard industrial pattern to protect database infrastructures against sudden traffic surges or hot-key cache expiration.
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/gophini/coalesce"
)
// Define your favorite local or distributed cache structure
type MemoryCache struct {
// e.g., sync.Map or LRU cache
}
func (m *MemoryCache) Get(key string) (any, bool) { return nil, false }
func (m *MemoryCache) Add(key string, value any) {}
func main() {
// Initialize the group barrier with an underlying context and optional cache tier
group := coalesce.NewGroup[string, string](
coalesce.WithGroupContext[string, string](context.Background()),
coalesce.WithGroupTaskAutoCancel[string, string](true),
coalesce.WithGroupCache[string, string](&MemoryCache{}), // your cache instance
)
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
// Heavy database operation simulation
fetchDB := func(bgCtx context.Context) (string, error) {
time.Sleep(100 * time.Millisecond) // Simulate slow query
return "user_profile_data", nil
}
// 100 concurrent goroutines calling this simultaneously will trigger the DB exactly ONCE.
val, shared, err := group.Get(ctx, "user_id_101", fetchDB)
if err != nil {
log.Fatalf("failed to fetch: %v", err)
}
// shared = true for all concurrent waiters, false only for the definitive winner.
fmt.Printf("Data: %s, Shared: %t\n", val, shared)
}
2. Safeguarding Global Resource Initialization (coalesce.Task)
When you need a single object (e.g., an encrypted configuration manager) to be safely instantiated on-demand under severe startup traffic.
package main
import (
"context"
"fmt"
"time"
"github.com/gophini/coalesce"
)
func main() {
// Enable AutoCancel so if all client HTTP requests abort, the background setup halts instantly.
task := coalesce.NewTask[string](
coalesce.WithTaskAutoCancel[string](true),
)
setupResource := func(ctx context.Context) (string, error) {
time.Sleep(500 * time.Millisecond)
return "global_connection_established", nil
}
// Blocks gracefully and returns exactly when setup completes
res, shared, _ := task.Get(context.Background(), setupResource)
fmt.Printf("Resource: %s, Shared: %t\n", res, shared)
}
🛠️ Performance & Concurrency Safety
coalesce leverages Go's native memory barriers (sync.RWMutex Double-Check locking and sync/atomic fences) to minimize CPU cache-line bouncing.
To test concurrency safety under heavy data-race analysis, run:
go test -v -race -count=5 ./...
📄 License
Distributed under the MIT License. See LICENSE for more information.
Documentation
¶
Index ¶
- type Cache
- type Group
- type Option
- func WithGroupCache[K comparable, V any](cache Cache[K, V]) Option[*groupOptions[K, V]]
- func WithGroupContext[K comparable, V any](ctx context.Context) Option[*groupOptions[K, V]]
- func WithGroupTaskAutoCancel[K comparable, V any](enabled bool) Option[*groupOptions[K, V]]
- func WithTaskAutoCancel[T any](enabled bool) Option[*taskOptions[T]]
- func WithTaskContext[T any](ctx context.Context) Option[*taskOptions[T]]
- func WithTaskDone[T any](done func(val T, err error)) Option[*taskOptions[T]]
- type Task
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Cache ¶
Cache defines the minimalistic storage blueprint required by the Group component. It allows seamless, out-of-the-box integration with memory caches (e.g., LRU) or remote systems (e.g., Redis).
type Group ¶
type Group[K comparable, V any] struct { // contains filtered or unexported fields }
Group manages concurrent read/write coalescing across a keyspace. It leverages localized Task[V] generations to isolate hot-key synchronization.
type Option ¶
type Option[T any] interface { // contains filtered or unexported methods }
func WithGroupCache ¶
func WithGroupCache[K comparable, V any](cache Cache[K, V]) Option[*groupOptions[K, V]]
WithGroupCache configures the secondary persistence or eviction tier (e.g., an LRU cache).
func WithGroupContext ¶
func WithGroupContext[K comparable, V any](ctx context.Context) Option[*groupOptions[K, V]]
WithGroupContext attaches a long-lived supervisor context to the cache broker.
func WithGroupTaskAutoCancel ¶
func WithGroupTaskAutoCancel[K comparable, V any](enabled bool) Option[*groupOptions[K, V]]
WithGroupTaskAutoCancel propagates the dynamic early-cleanup flag down to every individual key's sub-task execution.
func WithTaskAutoCancel ¶
WithAutoCancel activates the dynamic reference telemetry tracker. If all calling contexts timeout or abort, the active worker routine is terminated via surgical context cancellation, instantly saving CPU/IO resources and preventing ghost tasks from draining the system.
func WithTaskContext ¶
WithContext binds a long-lived supervisor context (e.g., Application or Plugin lifecycle) to the initialization lifecycle. This ensures that even if individual ephemeral client requests timeout or abort, the background bootstrap routine can safely run to completion using this stable context boundary, preventing corrupted partial states.
func WithTaskDone ¶
WithDone registers a thread-safe, asynchronous post-initialization callback function. It is fully type-safe and ideal for exporting connection metrics, telemetries, dynamic pooling, or firing alerting pipelines upon bootstrap failures.
type Task ¶
type Task[T any] struct { // contains filtered or unexported fields }
Task guards the lazy initialization of a heavy resource (e.g., global singletons, configurations, or connection pools) with request coalescing capabilities. It solves 4 major concurrent pain points: thundering herd protection, resilient lazy-loading, instant response to caller context cancellations, and automated teardown of orphaned generations.
func NewTask ¶
NewTask creates a ready-to-use, type-safe Task instance using generic functional options.
func (*Task[T]) Get ¶
Get ensures that the initialization function f is called exactly once as long as it succeeds. It returns (value, shared, error). CRITICAL SEMANTIC: shared is false ONLY when the specific caller won the race and successfully completed the initialization. For all concurrent waiters, aborted requests, or failed generations, shared returns true to strongly protect upper cache layers (e.g., Redis/LRU) from duplicate/bad writes.
Source Files