diff --git a/caddystorage/locker.go b/caddystorage/locker.go new file mode 100644 index 00000000..ecacf633 --- /dev/null +++ b/caddystorage/locker.go @@ -0,0 +1,172 @@ +package caddystorage + +import ( + "context" + "errors" + "fmt" + "log/slog" + "slices" + "time" + + "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/distlock" +) + +const lockResourcePrefix = "caddy_storage:" + +// Lock acquires an automatically renewed distributed lock and waits for the local store to catch up with versions +// observed on responding machines. Writers using the same lock can then read locally. Unavailable machines may have +// writes that this wait does not cover, and reads outside a lock remain eventually consistent. +func (s *Storage) Lock(ctx context.Context, name string) (err error) { + if name == "" { + return errors.New("lock name is empty") + } + s.locksMu.Lock() + if s.locks == nil { + s.locksMu.Unlock() + return errors.New("storage is closed") + } + s.locksMu.Unlock() + + // Stop acquisition retries when Caddy unloads the module, even if the caller's context is still active. + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + stopOnCleanup := context.AfterFunc(s.ctx, func() { + cancel(context.Cause(s.ctx)) + }) + defer stopOnCleanup() + + log := s.log.With("lock", name) + started := time.Now() + stage := "acquire_lease" + log.Debug("acquiring lock", "lock_ttl", time.Duration(s.LockTTL)) + defer func() { + if err != nil { + log.Debug("failed to acquire lock", + "stage", stage, "duration", time.Since(started), "error", err) + } + }() + + lease, err := s.locker.Acquire(ctx, lockResourcePrefix+name) + if err != nil { + return fmt.Errorf("acquire lock '%s': %w", name, err) + } + log.Debug("lock lease acquired", "duration", time.Since(started)) + // Keep observing after Lock returns so lease loss during protected work remains visible. + context.AfterFunc(lease.Context(), func() { + if cause := context.Cause(lease.Context()); errors.Is(cause, distlock.ErrLeaseLost) { + log.Error("lock lease lost", "error", cause) + } + }) + + defer func() { + if err == nil { + return + } + // Capture lease loss before Release cancels the lease context itself. + err = errors.Join(err, context.Cause(lease.Context())) + err = fmt.Errorf("acquire lock '%s': %w", name, + errors.Join(err, s.releaseLock(ctx, name, lease, "failed acquisition"))) + }() + + stopOnLeaseLoss := context.AfterFunc(lease.Context(), func() { + cancel(context.Cause(lease.Context())) + }) + defer stopOnLeaseLoss() + + stage = "collect_store_versions" + version, machines, err := s.clusterStoreVersion(ctx, log) + if err != nil { + return err + } + stage = "wait_for_replication" + waitStarted := time.Now() + log.Debug("waiting for local store replication", "machine_names", machines, "store_version", version) + if err := s.client.WaitForStoreVersion(ctx, version); err != nil { + return fmt.Errorf("wait for local store replication: %w", err) + } + log.Debug("local store replication complete", "duration", time.Since(waitStarted)) + + stage = "register_lock" + s.locksMu.Lock() + defer s.locksMu.Unlock() + if err := context.Cause(ctx); err != nil { + return err + } + if err := context.Cause(lease.Context()); err != nil { + return err + } + if s.locks == nil { + // A lease obtained during cleanup must be released instead of reopening the module's lock map. + return errors.New("storage is closed") + } + if _, exists := s.locks[name]; exists { + return errors.New("lock is already held by this storage instance") + } + s.locks[name] = lease + + log.Debug("lock acquired", "duration", time.Since(started)) + return nil +} + +// clusterStoreVersion returns the per-actor maximum store versions from responding machines and their names. +func (s *Storage) clusterStoreVersion(ctx context.Context, log *slog.Logger) (map[string]uint64, []string, error) { + ctx, cancel := context.WithTimeout(ctx, distlock.DefaultMaxNodeCallTimeout) + defer cancel() + resp, err := s.client.MachineClient.InspectMachine(client.ProxyMachinesContext(ctx, nil), nil) + if err != nil { + return nil, nil, fmt.Errorf("inspect machines for store versions: %w", err) + } + + maxVersion := make(map[string]uint64) + machines := make([]string, 0, len(resp.Machines)) + for _, m := range resp.Machines { + if m.Metadata.Error != "" { + log.Debug("skipping machine when collecting store versions", + "id", m.Metadata.MachineId, "name", m.Metadata.MachineName, "error", m.Metadata.Error) + continue + } + machines = append(machines, m.Metadata.MachineName) + for actor, v := range m.StoreVersion { + maxVersion[actor] = max(maxVersion[actor], v) + } + } + slices.Sort(machines) + return maxVersion, machines, nil +} + +// Unlock releases a previously acquired distributed lock. +func (s *Storage) Unlock(ctx context.Context, name string) error { + s.locksMu.Lock() + lease, exists := s.locks[name] + if exists { + delete(s.locks, name) + } + s.locksMu.Unlock() + if !exists { + return fmt.Errorf("lock '%s' is not held by this storage instance", name) + } + // Release stops renewal even on error. Any nodes that cannot be reached will let the lease expire. + if err := s.releaseLock(ctx, name, lease, "unlock"); err != nil { + return fmt.Errorf("release lock '%s': %w", name, err) + } + + return nil +} + +// releaseLock logs releases consistently across unlock, failed acquisition, and module cleanup. +func (s *Storage) releaseLock(ctx context.Context, name string, lease *distlock.Lease, reason string) error { + // Unlock and rollback must attempt node cleanup even if the caller or module has already been cancelled. + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), distlock.DefaultMaxNodeCallTimeout) + defer cancel() + + log := s.log.With("lock", name, "reason", reason) + started := time.Now() + log.Debug("releasing lock") + if err := lease.Release(ctx); err != nil { + log.Debug("failed to release lock", "duration", time.Since(started), "error", err) + return err + } + log.Debug("lock released", "duration", time.Since(started)) + return nil +} diff --git a/caddystorage/module.go b/caddystorage/module.go new file mode 100644 index 00000000..ab9ce17c --- /dev/null +++ b/caddystorage/module.go @@ -0,0 +1,205 @@ +// Package caddystorage provides Caddy storage backed by an Uncloud cluster. +package caddystorage + +import ( + "context" + "errors" + "fmt" + "log/slog" + "sync" + "time" + + "github.com/caddyserver/caddy/v2" + "github.com/caddyserver/caddy/v2/caddyconfig/caddyfile" + "github.com/caddyserver/certmagic" + "github.com/psviderski/uncloud/pkg/client" + "github.com/psviderski/uncloud/pkg/client/connector" + "github.com/psviderski/uncloud/pkg/distlock" +) + +const ( + // ModuleID is the Caddy module ID for Uncloud storage. + ModuleID = "caddy.storage.uncloud" + // DefaultSocketPath is the default path to the Uncloud API socket. + DefaultSocketPath = "/run/uncloud/uncloud.sock" + // DefaultLockTTL is the default duration of a distributed lock lease. + DefaultLockTTL = 20 * time.Second +) + +func init() { + caddy.RegisterModule(new(Storage)) +} + +// Storage implements a Caddy storage backend that uses an Uncloud cluster to store assets such as TLS certificates. +type Storage struct { + // Socket is the path to the Uncloud API socket. + // Defaults to /run/uncloud/uncloud.sock when not set. + Socket string `json:"socket,omitempty"` + // LockTTL is the duration of a distributed lock after which it expires if not renewed. Locks renew automatically + // until unlocked. If an instance crashes or cannot renew, expiry allows another instance to acquire the stale lock. + // Longer durations tolerate longer interruptions but delay recovery after a crash. Normal unlocks release the lock + // immediately. + // Defaults to 20 seconds when not set. + LockTTL caddy.Duration `json:"lock_ttl,omitempty"` + + client *client.Client + locker *distlock.Locker + log *slog.Logger + + locksMu sync.Mutex + locks map[string]*distlock.Lease +} + +// CaddyModule returns the Caddy module information. +func (*Storage) CaddyModule() caddy.ModuleInfo { + return caddy.ModuleInfo{ + ID: ModuleID, + New: func() caddy.Module { return new(Storage) }, + } +} + +// Provision connects the storage to the local Uncloud API and initialises the distributed locker. +func (s *Storage) Provision(ctx caddy.Context) error { + s.log = ctx.Slogger() + + if s.Socket == "" { + s.Socket = DefaultSocketPath + } + if s.LockTTL == 0 { + s.LockTTL = caddy.Duration(DefaultLockTTL) + } + if s.LockTTL < 0 { + return errors.New("lock_ttl must be positive") + } + + cli, err := client.New(ctx, connector.NewUnixConnector(s.Socket)) + if err != nil { + return fmt.Errorf("connect to Uncloud API: %w", err) + } + + locker, err := cli.NewLocker(distlock.Config{ + LeaseDuration: time.Duration(s.LockTTL), + }) + if err != nil { + _ = cli.Close() + return fmt.Errorf("create distributed locker: %w", err) + } + + s.client = cli + s.locker = locker + s.locks = make(map[string]*distlock.Lease) + + s.log.Debug("module provisioned", "socket", s.Socket, "lock_ttl", time.Duration(s.LockTTL)) + return nil +} + +// Cleanup releases active locks and closes the Uncloud API connection. +func (s *Storage) Cleanup() (err error) { + s.locksMu.Lock() + locks := s.locks + s.locks = nil + s.locksMu.Unlock() + + started := time.Now() + s.log.Debug("cleaning up module", "locks", len(locks)) + defer func() { + if err != nil { + s.log.Debug("failed to clean up module", "duration", time.Since(started), "error", err) + } else { + s.log.Debug("module cleanup complete", "duration", time.Since(started)) + } + }() + + ctx, cancel := context.WithTimeout(context.Background(), distlock.DefaultMaxNodeCallTimeout) + defer cancel() + + errCh := make(chan error, len(locks)) + var wg sync.WaitGroup + for name, lease := range locks { + wg.Go(func() { + if err := s.releaseLock(ctx, name, lease, "cleanup"); err != nil { + errCh <- fmt.Errorf("release lock '%s': %w", name, err) + } + }) + } + wg.Wait() + close(errCh) + + errs := make([]error, 0, len(errCh)+1) + for err := range errCh { + errs = append(errs, err) + } + + if s.client != nil { + errs = append(errs, s.client.Close()) + s.client = nil + s.locker = nil + } + + return errors.Join(errs...) +} + +// CertMagicStorage returns the provisioned CertMagic storage implementation. +func (s *Storage) CertMagicStorage() (certmagic.Storage, error) { + return s, nil +} + +// UnmarshalCaddyfile configures Uncloud storage from the Caddyfile global storage block. +// +// { +// storage uncloud { +// socket /run/uncloud/uncloud.sock +// lock_ttl 20s +// } +// } +func (s *Storage) UnmarshalCaddyfile(d *caddyfile.Dispenser) error { + d.Next() // Skip the module name 'uncloud'. + // Reject inline arguments. NextArg leaves an opening brace for NextBlock. + if d.NextArg() { + return d.ArgErr() + } + + // Read the optional options block, skipping its surrounding braces. + for d.NextBlock(0) { + switch d.Val() { + case "socket": + // Require a socket path on the same line as 'socket' option. + if !d.NextArg() { + return d.ArgErr() + } + s.Socket = d.Val() + // Reject extra arguments after the socket path. + if d.NextArg() { + return d.ArgErr() + } + case "lock_ttl": + if !d.NextArg() { + return d.ArgErr() + } + ttl, err := caddy.ParseDuration(d.Val()) + if err != nil { + return d.Errf("invalid lock_ttl '%s': %v", d.Val(), err) + } + if ttl <= 0 { + return d.Err("lock_ttl must be positive") + } + if d.NextArg() { + return d.ArgErr() + } + s.LockTTL = caddy.Duration(ttl) + default: + return d.Errf("unknown uncloud storage option: '%s'", d.Val()) + } + } + + return nil +} + +var ( + _ caddy.Module = (*Storage)(nil) + _ caddy.Provisioner = (*Storage)(nil) + _ caddy.CleanerUpper = (*Storage)(nil) + _ caddy.StorageConverter = (*Storage)(nil) + _ caddyfile.Unmarshaler = (*Storage)(nil) + _ certmagic.Storage = (*Storage)(nil) +) diff --git a/caddystorage/storage.go b/caddystorage/storage.go new file mode 100644 index 00000000..501e59be --- /dev/null +++ b/caddystorage/storage.go @@ -0,0 +1,84 @@ +package caddystorage + +import ( + "context" + "fmt" + "io/fs" + + "github.com/caddyserver/certmagic" + "github.com/psviderski/uncloud/api/pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Store writes a value to the cluster store. +func (s *Storage) Store(ctx context.Context, key string, value []byte) error { + if _, err := s.client.CaddyStorage.Store(ctx, &pb.StoreCaddyStorageRequest{Key: key, Value: value}); err != nil { + return storageError("store", key, err) + } + return nil +} + +// Load reads a value from the cluster store. +func (s *Storage) Load(ctx context.Context, key string) ([]byte, error) { + resp, err := s.client.CaddyStorage.Load(ctx, &pb.LoadCaddyStorageRequest{Key: key}) + if err != nil { + return nil, storageError("load", key, err) + } + return resp.Value, nil +} + +// Delete removes a key and its descendants from the cluster store. +func (s *Storage) Delete(ctx context.Context, key string) error { + if _, err := s.client.CaddyStorage.Delete(ctx, &pb.DeleteCaddyStorageRequest{Key: key}); err != nil { + return storageError("delete", key, err) + } + return nil +} + +// Exists reports whether a key exists in the cluster store. +func (s *Storage) Exists(ctx context.Context, key string) bool { + _, err := s.Stat(ctx, key) + return err == nil +} + +// List returns keys under prefix from the cluster store. +func (s *Storage) List(ctx context.Context, prefix string, recursive bool) ([]string, error) { + resp, err := s.client.CaddyStorage.List(ctx, &pb.ListCaddyStorageRequest{ + Prefix: prefix, + Recursive: recursive, + }) + if err != nil { + return nil, storageError("list", prefix, err) + } + return resp.Keys, nil +} + +// Stat returns information about a key in the cluster store. +func (s *Storage) Stat(ctx context.Context, key string) (certmagic.KeyInfo, error) { + resp, err := s.client.CaddyStorage.Stat(ctx, &pb.StatCaddyStorageRequest{Key: key}) + if err != nil { + return certmagic.KeyInfo{}, storageError("stat", key, err) + } + + info := certmagic.KeyInfo{ + Key: resp.Key, + Size: resp.Size, + IsTerminal: resp.IsTerminal, + } + if resp.UpdatedAt != nil { + if err := resp.UpdatedAt.CheckValid(); err != nil { + return certmagic.KeyInfo{}, storageError("stat", key, fmt.Errorf("invalid updated_at timestamp: %w", err)) + } + info.Modified = resp.UpdatedAt.AsTime() + } + return info, nil +} + +func storageError(operation, key string, err error) error { + if status.Code(err) == codes.NotFound { + err = fs.ErrNotExist + } + // TODO: figure out what the appropriate way of calling the storage in caddy logs is, e.g. include uncloud or redundant if logger includes the name? + return fmt.Errorf("%s Caddy storage key '%s': %w", operation, key, err) +}