A lightweight Go package for distributed command execution and coordination across services
Note
This version is written by Claude after explaining the requirements and providing some code that I used somewhere else. I haven't tested it extensively, please use with caution. I will update once I get the chance to test it thoroughly.
cnc-go lets you:
-
⚡ Register commands with custom handlers
-
📡 Trigger commands locally or across a cluster
-
🔌 Propagate events using pluggable Pub/Sub backends (Redis/Valkey included, extendable to NATS, Kafka, etc.)
-
🗄️ Ensure synchronized actions across pods/nodes
-
🛠️ Gracefully start, run, and shutdown CNC workers
In distributed systems, some actions — like invalidating caches, reloading configs, or resetting feature flags — must be executed everywhere.
Instead of building ad-hoc RPCs or reinventing messaging, cnc-go provides a simple abstraction: publish a command once and have every node execute it reliably.
-
🔌 Pluggable Pub/Sub transport (Redis included, extendable to NATS/Kafka/etc.)
-
🧩 Simple API for registering and triggering commands
-
🛡️ Typed errors for safe error handling (
ErrHandlerNotFound,ErrHandlerAlreadyExists, …) -
📦 Clean, idiomatic Go design (no external dependencies beyond the transport driver)
-
✅ Unit tested & extensible
go get github.com/TheAlpha16/cnc-gopackage main import ( "context" "fmt" "log" "time" "github.com/TheAlpha16/cnc-go" "github.com/valkey-io/valkey-go" ) func main() { // Create CNC instance with Valkey transport cncInstance, err := cnc.NewCNCWithValkeyAddress("localhost:6379", "my-commands", []valkey.ClientOption{}) if err != nil { log.Fatalf("Failed to create CNC: %v", err) } defer cncInstance.Shutdown() // Register a simple command handler err = cncInstance.RegisterHandler("hello", func(ctx context.Context, cmd cnc.Command) error { name := "World" if n, ok := cmd.Parameters["name"].(string); ok { name = n } fmt.Printf("Hello, %s!\n", name) return nil }) if err != nil { log.Fatalf("Failed to register handler: %v", err) } // Start the CNC instance ctx := context.Background() if err := cncInstance.Start(ctx); err != nil { log.Fatalf("Failed to start CNC: %v", err) } fmt.Println("CNC started! Transport is now listening for commands...") // Trigger a command after a short delay time.Sleep(1 * time.Second) command := cnc.Command{ Name: "hello", Parameters: map[string]any{ "name": "CNC User", }, } if err := cncInstance.TriggerCommand(ctx, command); err != nil { log.Printf("Failed to trigger command: %v", err) } // Keep the program running for a bit to see the result time.Sleep(3 * time.Second) fmt.Println("Quick start example completed!") }The main interface for command and control operations:
type CNC interface { RegisterHandler(commandName CommandName, handler Handler) error TriggerCommand(ctx context.Context, command Command) error Start(ctx context.Context) error Shutdown() error IsRunning() bool }Pluggable transport layer for message passing:
type Transport interface { Publish(ctx context.Context, command Command) error Subscribe(ctx context.Context) error Messages() <-chan Command // Returns channel of received commands Close() error IsConnected() bool }client, err := cnc.NewValkeyClient("localhost:6379") if err != nil { log.Fatal(err) } cncInstance := cnc.NewCNCWithValkey(client, "command-channel")cncInstance, err := cnc.NewCNCWithValkeyAddress("localhost:6379", "command-channel") if err != nil { log.Fatal(err) }transport := NewMyCustomTransport() cncInstance := cnc.NewCNC(transport)type Command struct { Name CommandName `json:"type"` Parameters map[string]any `json:"parameters,omitempty"` } type Handler func(ctx context.Context, command Command) errorThe package provides typed errors for better error handling:
var ( ErrInvalidCommand = errors.New("invalid command") ErrHandlerAlreadyExists = errors.New("handler already exists") ErrHandlerNotFound = errors.New("handler not found") ErrPublishFailed = errors.New("failed to publish command") ErrSubscribeFailed = errors.New("failed to subscribe to channel") ErrTransportNotConnected = errors.New("transport not connected") ErrTransportClosed = errors.New("transport is closed") ErrCNCNotStarted = errors.New("cnc instance not started") ErrCNCAlreadyStarted = errors.New("cnc instance already started") )To implement a custom transport (e.g., for NATS, Kafka), implement the Transport interface:
type MyCustomTransport struct { msgChan chan Command // Your other transport fields } func (t *MyCustomTransport) Publish(ctx context.Context, command Command) error { // Implement publishing logic return nil } func (t *MyCustomTransport) Subscribe(ctx context.Context) error { // Start listening for messages and feed them to msgChan return nil } func (t *MyCustomTransport) Messages() <-chan Command { // Return the channel that receives commands return t.msgChan } func (t *MyCustomTransport) Close() error { // Implement cleanup logic close(t.msgChan) return nil } func (t *MyCustomTransport) IsConnected() bool { // Return connection status return true }Run the tests:
go test ./...Run tests with coverage:
go test -cover ./...- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add some amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
This project is licensed under the MIT License - see the LICENSE file for details.
The package follows a clean separation of concerns:
- Transport Layer: Pure message passing (publish/subscribe) with channels
- Registry: Maps command names to handler functions
- CNC Core: Orchestrates transport and registry, manages lifecycle
Commands ──┐ ▼ ┌─────────────┐ ┌──────────────┐ ┌────────────┐ │ Transport │───▶│ Channel │───▶│ CNC Core │ │ │ │ <-chan Cmd │ │ │ └─────────────┘ └──────────────┘ └────────────┘ │ ▼ ┌─────────────┐ │ Registry │ │ (Handlers) │ └─────────────┘ If you have any questions or need help:
- 📧 Open an issue on GitHub
- 💬 Start a discussion in the repository
- ⭐ Star the repo if you find it useful!