From 7a7a313426f5ebd42ba50d494bc32805e3ee94cb Mon Sep 17 00:00:00 2001 From: Pasha Sviderski Date: Thu, 27 Aug 2026 12:27:54 +1000 Subject: [PATCH] feat(distlock): add gRPC Lease server that adapts a machine-local Store --- internal/distlock/server.go | 98 +++++++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) create mode 100644 internal/distlock/server.go diff --git a/internal/distlock/server.go b/internal/distlock/server.go new file mode 100644 index 00000000..30026ea1 --- /dev/null +++ b/internal/distlock/server.go @@ -0,0 +1,98 @@ +package distlock + +import ( + "context" + "fmt" + "time" + + "github.com/psviderski/uncloud/internal/machine/api/pb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" +) + +// Server adapts a machine-local Store to the Lease gRPC service. +type Server struct { + pb.UnimplementedLeaseServer + store Store +} + +// NewServer creates a machine-local lease server. +func NewServer(store Store) *Server { + return &Server{store: store} +} + +// Acquire creates a lease when the resource has no unexpired lease. +func (s *Server) Acquire(ctx context.Context, req *pb.AcquireLeaseRequest) (*pb.AcquireLeaseResponse, error) { + ttl, err := validateLeaseRequest(req.Resource, req.Token, req.Ttl) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + + acquired, err := s.store.Acquire(ctx, req.Resource, req.Token, ttl) + if err != nil { + return nil, storeStatusError(ctx, "acquire lease", err) + } + return &pb.AcquireLeaseResponse{Acquired: acquired}, nil +} + +// Renew extends an unexpired lease when its ownership token matches. +func (s *Server) Renew(ctx context.Context, req *pb.RenewLeaseRequest) (*pb.RenewLeaseResponse, error) { + ttl, err := validateLeaseRequest(req.Resource, req.Token, req.Ttl) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + + renewed, err := s.store.Renew(ctx, req.Resource, req.Token, ttl) + if err != nil { + return nil, storeStatusError(ctx, "renew lease", err) + } + return &pb.RenewLeaseResponse{Renewed: renewed}, nil +} + +// Release removes an unexpired lease when its ownership token matches. +func (s *Server) Release(ctx context.Context, req *pb.ReleaseLeaseRequest) (*pb.ReleaseLeaseResponse, error) { + if err := validateResourceToken(req.Resource, req.Token); err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + + released, err := s.store.Release(ctx, req.Resource, req.Token) + if err != nil { + return nil, storeStatusError(ctx, "release lease", err) + } + return &pb.ReleaseLeaseResponse{Released: released}, nil +} + +func validateLeaseRequest(resource string, token []byte, ttl *durationpb.Duration) (time.Duration, error) { + if err := validateResourceToken(resource, token); err != nil { + return 0, err + } + if ttl == nil { + return 0, fmt.Errorf("TTL is not set") + } + if err := ttl.CheckValid(); err != nil { + return 0, fmt.Errorf("invalid TTL: %w", err) + } + duration := ttl.AsDuration() + if duration <= 0 { + return 0, fmt.Errorf("TTL must be positive") + } + return duration, nil +} + +func validateResourceToken(resource string, token []byte) error { + if resource == "" { + return fmt.Errorf("resource is empty") + } + if len(token) == 0 { + return fmt.Errorf("token is empty") + } + return nil +} + +func storeStatusError(ctx context.Context, operation string, err error) error { + if ctxErr := ctx.Err(); ctxErr != nil { + return status.FromContextError(ctxErr).Err() + } + return status.Error(codes.Internal, fmt.Sprintf("%s: %v", operation, err)) +}