From 1861b688e9e0ba6ffd04234e7829e4aec1c335c7 Mon Sep 17 00:00:00 2001 From: maml Date: Sat, 18 Apr 2026 13:10:47 -0400 Subject: [PATCH] adminrpc: add per-service timeout field for TTL caveat support Admin-API-registered services previously had Timeout=0 after the mergeServicesFromDB merge, because the proto's Service message lacked a timeout field. As a result staticServiceLimiter.ServiceTimeouts never added a _valid_until caveat for admin-API-sourced services, silently disabling the TTL expiry mechanism documented in sample-conf.yaml. This commit adds int64 timeout to: - Service - CreateServiceRequest - UpdateServiceRequest and wires it through: - aperturedb services table (schema + migration 000008) - aperturedb/sqlc: models, query, generated services.sql.go - aperturedb/services.go ServiceParams - admin/server.go CreateService, UpdateService, ListServices - aperture.go mergeServicesFromDB (copy Timeout to proxy.Service) No changes to staticServiceLimiter.ServiceTimeouts or the mint pipeline -- they already check proxyService.Timeout > 0. This commit surfaces the existing mechanism through the admin API surface. Other per-service fields (capabilities, constraints, ratelimits, authwhitelistpaths, headers, rewrite, tlscertpath) have the same gap and are intentionally deferred to follow-up PRs to keep this change focused. --- admin/server.go | 24 ++++ admin/server_test.go | 109 ++++++++++++++++++ adminrpc/admin.pb.go | 53 +++++++-- adminrpc/admin.proto | 12 ++ adminrpc/admin.swagger.json | 15 +++ adminrpc/admin_grpc.pb.go | 90 ++++++++++----- aperture.go | 18 +-- aperturedb/services.go | 2 + aperturedb/services_test.go | 64 ++++++++++ .../000008_services_timeout.down.sql | 1 + .../migrations/000008_services_timeout.up.sql | 1 + aperturedb/sqlc/models.go | 1 + aperturedb/sqlc/queries/services.sql | 5 +- aperturedb/sqlc/services.sql.go | 10 +- docs/admin-api.md | 1 + services.go | 46 +++++--- services_test.go | 104 +++++++++++++++++ 17 files changed, 495 insertions(+), 61 deletions(-) create mode 100644 aperturedb/sqlc/migrations/000008_services_timeout.down.sql create mode 100644 aperturedb/sqlc/migrations/000008_services_timeout.up.sql diff --git a/admin/server.go b/admin/server.go index 04187eec..23f192ed 100644 --- a/admin/server.go +++ b/admin/server.go @@ -174,6 +174,7 @@ func (s *Server) ListServices(_ context.Context, Price: svc.Price, Auth: string(svc.Auth), AuthScheme: stringToAuthScheme(svc.AuthScheme), + Timeout: svc.Timeout, }) } @@ -278,6 +279,12 @@ func (s *Server) CreateService(ctx context.Context, authScheme := authSchemeToString(req.AuthScheme) + if req.Timeout < 0 { + return nil, status.Error( + codes.InvalidArgument, "timeout must be >= 0", + ) + } + newSvc := &proxy.Service{ Name: req.Name, Address: req.Address, @@ -286,6 +293,7 @@ func (s *Server) CreateService(ctx context.Context, PathRegexp: req.PathRegexp, Price: req.Price, AuthScheme: authScheme, + Timeout: req.Timeout, } if normalizedAuth != "" { newSvc.Auth = auth.Level(normalizedAuth) @@ -312,6 +320,7 @@ func (s *Server) CreateService(ctx context.Context, Auth: string(newSvc.Auth), AuthScheme: newSvc.AuthScheme, Price: newSvc.Price, + Timeout: newSvc.Timeout, }, ); err != nil { log.Errorf("Error persisting service: %v", err) @@ -327,6 +336,7 @@ func (s *Server) CreateService(ctx context.Context, Price: newSvc.Price, Auth: string(newSvc.Auth), AuthScheme: stringToAuthScheme(newSvc.AuthScheme), + Timeout: newSvc.Timeout, }, nil } @@ -438,6 +448,18 @@ func (s *Server) UpdateService(ctx context.Context, updated.AuthScheme = authSchemeToString(*req.AuthScheme) } + // Only apply timeout when explicitly set. Using `optional` lets + // callers distinguish "reset to 0 (no expiry)" from "leave as-is". + if req.Timeout != nil { + if req.GetTimeout() < 0 { + return nil, status.Error( + codes.InvalidArgument, + "timeout must be >= 0", + ) + } + updated.Timeout = req.GetTimeout() + } + // Replace the pointer in the slice with the updated copy. for i, svc := range services { if svc.Name == req.Name { @@ -466,6 +488,7 @@ func (s *Server) UpdateService(ctx context.Context, Auth: string(updated.Auth), AuthScheme: updated.AuthScheme, Price: updated.Price, + Timeout: updated.Timeout, }, ); err != nil { log.Errorf("Error persisting updated service: %v", @@ -482,6 +505,7 @@ func (s *Server) UpdateService(ctx context.Context, Price: updated.Price, Auth: string(updated.Auth), AuthScheme: stringToAuthScheme(updated.AuthScheme), + Timeout: updated.Timeout, }, nil } diff --git a/admin/server_test.go b/admin/server_test.go index e34588e4..cbd9ddc0 100644 --- a/admin/server_test.go +++ b/admin/server_test.go @@ -302,6 +302,115 @@ func TestUpdateServiceRejectsInvalidAuth(t *testing.T) { require.Contains(t, err.Error(), "invalid freebie count") } +func TestCreateServiceWithTimeout(t *testing.T) { + t.Parallel() + + s := newTestServer() + + svc, err := s.CreateService(context.Background(), + &adminrpc.CreateServiceRequest{ + Name: "timed-svc", + Address: "localhost:9999", + PathRegexp: "^/api/timed/.*", + Price: 100, + Timeout: 60, + }, + ) + require.NoError(t, err) + require.Equal(t, int64(60), svc.Timeout) + + // Verify timeout is returned by ListServices. + resp, err := s.ListServices( + context.Background(), &adminrpc.ListServicesRequest{}, + ) + require.NoError(t, err) + var found *adminrpc.Service + for _, s := range resp.Services { + if s.Name == "timed-svc" { + found = s + break + } + } + require.NotNil(t, found) + require.Equal(t, int64(60), found.Timeout) +} + +func TestUpdateServiceTimeout(t *testing.T) { + t.Parallel() + + s := newTestServer() + + timeout := int64(120) + svc, err := s.UpdateService(context.Background(), + &adminrpc.UpdateServiceRequest{ + Name: "test-svc", + Timeout: &timeout, + }, + ) + require.NoError(t, err) + require.Equal(t, int64(120), svc.Timeout) +} + +func TestUpdateServiceCanSetTimeoutToZero(t *testing.T) { + t.Parallel() + + s := newTestServer() + + // First set a non-zero timeout. + timeout := int64(60) + _, err := s.UpdateService(context.Background(), + &adminrpc.UpdateServiceRequest{ + Name: "test-svc", + Timeout: &timeout, + }, + ) + require.NoError(t, err) + + // Now reset to 0 (no expiry) via optional field. + zero := int64(0) + svc, err := s.UpdateService(context.Background(), + &adminrpc.UpdateServiceRequest{ + Name: "test-svc", + Timeout: &zero, + }, + ) + require.NoError(t, err) + require.Equal(t, int64(0), svc.Timeout) +} + +func TestCreateServiceRejectsNegativeTimeout(t *testing.T) { + t.Parallel() + + s := newTestServer() + + _, err := s.CreateService(context.Background(), + &adminrpc.CreateServiceRequest{ + Name: "bad-timeout-svc", + Address: "localhost:1234", + PathRegexp: "^/api/bad/.*", + Timeout: -1, + }, + ) + require.Error(t, err) + require.Contains(t, err.Error(), "timeout must be >= 0") +} + +func TestUpdateServiceRejectsNegativeTimeout(t *testing.T) { + t.Parallel() + + s := newTestServer() + + timeout := int64(-5) + _, err := s.UpdateService(context.Background(), + &adminrpc.UpdateServiceRequest{ + Name: "test-svc", + Timeout: &timeout, + }, + ) + require.Error(t, err) + require.Contains(t, err.Error(), "timeout must be >= 0") +} + func TestDeleteService(t *testing.T) { t.Parallel() diff --git a/adminrpc/admin.pb.go b/adminrpc/admin.pb.go index 5e1685cb..a5bb5cd7 100644 --- a/adminrpc/admin.pb.go +++ b/adminrpc/admin.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.10 -// protoc v3.21.12 +// protoc v7.34.0 // source: admin.proto package adminrpc @@ -372,7 +372,11 @@ type Service struct { Auth string `protobuf:"bytes,7,opt,name=auth,proto3" json:"auth,omitempty"` // auth_scheme specifies which payment auth scheme(s) are used for this // service. Defaults to AUTH_SCHEME_L402 for backwards compatibility. - AuthScheme AuthScheme `protobuf:"varint,8,opt,name=auth_scheme,json=authScheme,proto3,enum=adminrpc.AuthScheme" json:"auth_scheme,omitempty"` + AuthScheme AuthScheme `protobuf:"varint,8,opt,name=auth_scheme,json=authScheme,proto3,enum=adminrpc.AuthScheme" json:"auth_scheme,omitempty"` + // timeout is the per-service TTL in seconds. When non-zero, the L402 mint + // will include a _valid_until caveat so macaroons expire after this many + // seconds. A value of 0 means no expiry (no _valid_until caveat minted). + Timeout int64 `protobuf:"varint,9,opt,name=timeout,proto3" json:"timeout,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -463,6 +467,13 @@ func (x *Service) GetAuthScheme() AuthScheme { return AuthScheme_AUTH_SCHEME_L402 } +func (x *Service) GetTimeout() int64 { + if x != nil { + return x.Timeout + } + return 0 +} + type CreateServiceRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` @@ -474,7 +485,9 @@ type CreateServiceRequest struct { Auth string `protobuf:"bytes,7,opt,name=auth,proto3" json:"auth,omitempty"` // auth_scheme specifies which payment auth scheme(s) to use. Defaults to // AUTH_SCHEME_L402 if unset. - AuthScheme AuthScheme `protobuf:"varint,8,opt,name=auth_scheme,json=authScheme,proto3,enum=adminrpc.AuthScheme" json:"auth_scheme,omitempty"` + AuthScheme AuthScheme `protobuf:"varint,8,opt,name=auth_scheme,json=authScheme,proto3,enum=adminrpc.AuthScheme" json:"auth_scheme,omitempty"` + // timeout is the per-service TTL in seconds. See Service.timeout. + Timeout int64 `protobuf:"varint,9,opt,name=timeout,proto3" json:"timeout,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -565,6 +578,13 @@ func (x *CreateServiceRequest) GetAuthScheme() AuthScheme { return AuthScheme_AUTH_SCHEME_L402 } +func (x *CreateServiceRequest) GetTimeout() int64 { + if x != nil { + return x.Timeout + } + return 0 +} + type UpdateServiceRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` @@ -576,7 +596,10 @@ type UpdateServiceRequest struct { Auth string `protobuf:"bytes,7,opt,name=auth,proto3" json:"auth,omitempty"` // auth_scheme specifies which payment auth scheme(s) to use. When not // set, the existing auth_scheme is preserved (not reset to L402). - AuthScheme *AuthScheme `protobuf:"varint,8,opt,name=auth_scheme,json=authScheme,proto3,enum=adminrpc.AuthScheme,oneof" json:"auth_scheme,omitempty"` + AuthScheme *AuthScheme `protobuf:"varint,8,opt,name=auth_scheme,json=authScheme,proto3,enum=adminrpc.AuthScheme,oneof" json:"auth_scheme,omitempty"` + // timeout is the per-service TTL in seconds. See Service.timeout. When not + // set, the existing timeout is preserved. + Timeout *int64 `protobuf:"varint,9,opt,name=timeout,proto3,oneof" json:"timeout,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -667,6 +690,13 @@ func (x *UpdateServiceRequest) GetAuthScheme() AuthScheme { return AuthScheme_AUTH_SCHEME_L402 } +func (x *UpdateServiceRequest) GetTimeout() int64 { + if x != nil && x.Timeout != nil { + return *x.Timeout + } + return 0 +} + type DeleteServiceRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` @@ -1367,7 +1397,7 @@ const file_admin_proto_rawDesc = "" + "\x06status\x18\x01 \x01(\tR\x06status\"\x15\n" + "\x13ListServicesRequest\"E\n" + "\x14ListServicesResponse\x12-\n" + - "\bservices\x18\x01 \x03(\v2\x11.adminrpc.ServiceR\bservices\"\xf6\x01\n" + + "\bservices\x18\x01 \x03(\v2\x11.adminrpc.ServiceR\bservices\"\x90\x02\n" + "\aService\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x18\n" + "\aaddress\x18\x02 \x01(\tR\aaddress\x12\x1a\n" + @@ -1379,7 +1409,8 @@ const file_admin_proto_rawDesc = "" + "\x05price\x18\x06 \x01(\x03R\x05price\x12\x12\n" + "\x04auth\x18\a \x01(\tR\x04auth\x125\n" + "\vauth_scheme\x18\b \x01(\x0e2\x14.adminrpc.AuthSchemeR\n" + - "authScheme\"\x83\x02\n" + + "authScheme\x12\x18\n" + + "\atimeout\x18\t \x01(\x03R\atimeout\"\x9d\x02\n" + "\x14CreateServiceRequest\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x18\n" + "\aaddress\x18\x02 \x01(\tR\aaddress\x12\x1a\n" + @@ -1391,7 +1422,8 @@ const file_admin_proto_rawDesc = "" + "\x05price\x18\x06 \x01(\x03R\x05price\x12\x12\n" + "\x04auth\x18\a \x01(\tR\x04auth\x125\n" + "\vauth_scheme\x18\b \x01(\x0e2\x14.adminrpc.AuthSchemeR\n" + - "authScheme\"\xa7\x02\n" + + "authScheme\x12\x18\n" + + "\atimeout\x18\t \x01(\x03R\atimeout\"\xd2\x02\n" + "\x14UpdateServiceRequest\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x18\n" + "\aaddress\x18\x02 \x01(\tR\aaddress\x12\x1a\n" + @@ -1403,9 +1435,12 @@ const file_admin_proto_rawDesc = "" + "\x05price\x18\x06 \x01(\x03H\x00R\x05price\x88\x01\x01\x12\x12\n" + "\x04auth\x18\a \x01(\tR\x04auth\x12:\n" + "\vauth_scheme\x18\b \x01(\x0e2\x14.adminrpc.AuthSchemeH\x01R\n" + - "authScheme\x88\x01\x01B\b\n" + + "authScheme\x88\x01\x01\x12\x1d\n" + + "\atimeout\x18\t \x01(\x03H\x02R\atimeout\x88\x01\x01B\b\n" + "\x06_priceB\x0e\n" + - "\f_auth_scheme\"*\n" + + "\f_auth_schemeB\n" + + "\n" + + "\b_timeout\"*\n" + "\x14DeleteServiceRequest\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\"/\n" + "\x15DeleteServiceResponse\x12\x16\n" + diff --git a/adminrpc/admin.proto b/adminrpc/admin.proto index 0bedc4e0..5c248725 100644 --- a/adminrpc/admin.proto +++ b/adminrpc/admin.proto @@ -71,6 +71,11 @@ message Service { // auth_scheme specifies which payment auth scheme(s) are used for this // service. Defaults to AUTH_SCHEME_L402 for backwards compatibility. AuthScheme auth_scheme = 8; + + // timeout is the per-service TTL in seconds. When non-zero, the L402 mint + // will include a _valid_until caveat so macaroons expire after this many + // seconds. A value of 0 means no expiry (no _valid_until caveat minted). + int64 timeout = 9; } message CreateServiceRequest { @@ -85,6 +90,9 @@ message CreateServiceRequest { // auth_scheme specifies which payment auth scheme(s) to use. Defaults to // AUTH_SCHEME_L402 if unset. AuthScheme auth_scheme = 8; + + // timeout is the per-service TTL in seconds. See Service.timeout. + int64 timeout = 9; } message UpdateServiceRequest { @@ -99,6 +107,10 @@ message UpdateServiceRequest { // auth_scheme specifies which payment auth scheme(s) to use. When not // set, the existing auth_scheme is preserved (not reset to L402). optional AuthScheme auth_scheme = 8; + + // timeout is the per-service TTL in seconds. See Service.timeout. When not + // set, the existing timeout is preserved. + optional int64 timeout = 9; } message DeleteServiceRequest { string name = 1; } diff --git a/adminrpc/admin.swagger.json b/adminrpc/admin.swagger.json index af9ab8b1..0f5f4406 100644 --- a/adminrpc/admin.swagger.json +++ b/adminrpc/admin.swagger.json @@ -193,6 +193,11 @@ "auth_scheme": { "$ref": "#/definitions/adminrpcAuthScheme", "description": "auth_scheme specifies which payment auth scheme(s) to use. When not\nset, the existing auth_scheme is preserved (not reset to L402)." + }, + "timeout": { + "type": "string", + "format": "int64", + "description": "timeout is the per-service TTL in seconds. See Service.timeout. When not\nset, the existing timeout is preserved." } } } @@ -409,6 +414,11 @@ "auth_scheme": { "$ref": "#/definitions/adminrpcAuthScheme", "description": "auth_scheme specifies which payment auth scheme(s) to use. Defaults to\nAUTH_SCHEME_L402 if unset." + }, + "timeout": { + "type": "string", + "format": "int64", + "description": "timeout is the per-service TTL in seconds. See Service.timeout." } } }, @@ -554,6 +564,11 @@ "auth_scheme": { "$ref": "#/definitions/adminrpcAuthScheme", "description": "auth_scheme specifies which payment auth scheme(s) are used for this\nservice. Defaults to AUTH_SCHEME_L402 for backwards compatibility." + }, + "timeout": { + "type": "string", + "format": "int64", + "description": "timeout is the per-service TTL in seconds. When non-zero, the L402 mint\nwill include a _valid_until caveat so macaroons expire after this many\nseconds. A value of 0 means no expiry (no _valid_until caveat minted)." } } }, diff --git a/adminrpc/admin_grpc.pb.go b/adminrpc/admin_grpc.pb.go index a0570b70..7651ffdb 100644 --- a/adminrpc/admin_grpc.pb.go +++ b/adminrpc/admin_grpc.pb.go @@ -1,4 +1,8 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.5.1 +// - protoc v7.34.0 +// source: admin.proto package adminrpc @@ -11,8 +15,21 @@ import ( // This is a compile-time assertion to ensure that this generated file // is compatible with the grpc package it is being compiled against. -// Requires gRPC-Go v1.32.0 or later. -const _ = grpc.SupportPackageIsVersion7 +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + Admin_GetInfo_FullMethodName = "/adminrpc.Admin/GetInfo" + Admin_GetHealth_FullMethodName = "/adminrpc.Admin/GetHealth" + Admin_ListServices_FullMethodName = "/adminrpc.Admin/ListServices" + Admin_CreateService_FullMethodName = "/adminrpc.Admin/CreateService" + Admin_UpdateService_FullMethodName = "/adminrpc.Admin/UpdateService" + Admin_DeleteService_FullMethodName = "/adminrpc.Admin/DeleteService" + Admin_ListTransactions_FullMethodName = "/adminrpc.Admin/ListTransactions" + Admin_ListTokens_FullMethodName = "/adminrpc.Admin/ListTokens" + Admin_RevokeToken_FullMethodName = "/adminrpc.Admin/RevokeToken" + Admin_GetStats_FullMethodName = "/adminrpc.Admin/GetStats" +) // AdminClient is the client API for Admin service. // @@ -39,8 +56,9 @@ func NewAdminClient(cc grpc.ClientConnInterface) AdminClient { } func (c *adminClient) GetInfo(ctx context.Context, in *GetInfoRequest, opts ...grpc.CallOption) (*GetInfoResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetInfoResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/GetInfo", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_GetInfo_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -48,8 +66,9 @@ func (c *adminClient) GetInfo(ctx context.Context, in *GetInfoRequest, opts ...g } func (c *adminClient) GetHealth(ctx context.Context, in *GetHealthRequest, opts ...grpc.CallOption) (*GetHealthResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetHealthResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/GetHealth", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_GetHealth_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -57,8 +76,9 @@ func (c *adminClient) GetHealth(ctx context.Context, in *GetHealthRequest, opts } func (c *adminClient) ListServices(ctx context.Context, in *ListServicesRequest, opts ...grpc.CallOption) (*ListServicesResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListServicesResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/ListServices", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_ListServices_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -66,8 +86,9 @@ func (c *adminClient) ListServices(ctx context.Context, in *ListServicesRequest, } func (c *adminClient) CreateService(ctx context.Context, in *CreateServiceRequest, opts ...grpc.CallOption) (*Service, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(Service) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/CreateService", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_CreateService_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -75,8 +96,9 @@ func (c *adminClient) CreateService(ctx context.Context, in *CreateServiceReques } func (c *adminClient) UpdateService(ctx context.Context, in *UpdateServiceRequest, opts ...grpc.CallOption) (*Service, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(Service) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/UpdateService", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_UpdateService_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -84,8 +106,9 @@ func (c *adminClient) UpdateService(ctx context.Context, in *UpdateServiceReques } func (c *adminClient) DeleteService(ctx context.Context, in *DeleteServiceRequest, opts ...grpc.CallOption) (*DeleteServiceResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(DeleteServiceResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/DeleteService", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_DeleteService_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -93,8 +116,9 @@ func (c *adminClient) DeleteService(ctx context.Context, in *DeleteServiceReques } func (c *adminClient) ListTransactions(ctx context.Context, in *ListTransactionsRequest, opts ...grpc.CallOption) (*ListTransactionsResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListTransactionsResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/ListTransactions", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_ListTransactions_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -102,8 +126,9 @@ func (c *adminClient) ListTransactions(ctx context.Context, in *ListTransactions } func (c *adminClient) ListTokens(ctx context.Context, in *ListTokensRequest, opts ...grpc.CallOption) (*ListTokensResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(ListTokensResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/ListTokens", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_ListTokens_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -111,8 +136,9 @@ func (c *adminClient) ListTokens(ctx context.Context, in *ListTokensRequest, opt } func (c *adminClient) RevokeToken(ctx context.Context, in *RevokeTokenRequest, opts ...grpc.CallOption) (*RevokeTokenResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(RevokeTokenResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/RevokeToken", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_RevokeToken_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -120,8 +146,9 @@ func (c *adminClient) RevokeToken(ctx context.Context, in *RevokeTokenRequest, o } func (c *adminClient) GetStats(ctx context.Context, in *GetStatsRequest, opts ...grpc.CallOption) (*GetStatsResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) out := new(GetStatsResponse) - err := c.cc.Invoke(ctx, "/adminrpc.Admin/GetStats", in, out, opts...) + err := c.cc.Invoke(ctx, Admin_GetStats_FullMethodName, in, out, cOpts...) if err != nil { return nil, err } @@ -130,7 +157,7 @@ func (c *adminClient) GetStats(ctx context.Context, in *GetStatsRequest, opts .. // AdminServer is the server API for Admin service. // All implementations must embed UnimplementedAdminServer -// for forward compatibility +// for forward compatibility. type AdminServer interface { GetInfo(context.Context, *GetInfoRequest) (*GetInfoResponse, error) GetHealth(context.Context, *GetHealthRequest) (*GetHealthResponse, error) @@ -145,9 +172,12 @@ type AdminServer interface { mustEmbedUnimplementedAdminServer() } -// UnimplementedAdminServer must be embedded to have forward compatible implementations. -type UnimplementedAdminServer struct { -} +// UnimplementedAdminServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedAdminServer struct{} func (UnimplementedAdminServer) GetInfo(context.Context, *GetInfoRequest) (*GetInfoResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method GetInfo not implemented") @@ -180,6 +210,7 @@ func (UnimplementedAdminServer) GetStats(context.Context, *GetStatsRequest) (*Ge return nil, status.Errorf(codes.Unimplemented, "method GetStats not implemented") } func (UnimplementedAdminServer) mustEmbedUnimplementedAdminServer() {} +func (UnimplementedAdminServer) testEmbeddedByValue() {} // UnsafeAdminServer may be embedded to opt out of forward compatibility for this service. // Use of this interface is not recommended, as added methods to AdminServer will @@ -189,6 +220,13 @@ type UnsafeAdminServer interface { } func RegisterAdminServer(s grpc.ServiceRegistrar, srv AdminServer) { + // If the following call pancis, it indicates UnimplementedAdminServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } s.RegisterService(&Admin_ServiceDesc, srv) } @@ -202,7 +240,7 @@ func _Admin_GetInfo_Handler(srv interface{}, ctx context.Context, dec func(inter } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/GetInfo", + FullMethod: Admin_GetInfo_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).GetInfo(ctx, req.(*GetInfoRequest)) @@ -220,7 +258,7 @@ func _Admin_GetHealth_Handler(srv interface{}, ctx context.Context, dec func(int } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/GetHealth", + FullMethod: Admin_GetHealth_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).GetHealth(ctx, req.(*GetHealthRequest)) @@ -238,7 +276,7 @@ func _Admin_ListServices_Handler(srv interface{}, ctx context.Context, dec func( } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/ListServices", + FullMethod: Admin_ListServices_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).ListServices(ctx, req.(*ListServicesRequest)) @@ -256,7 +294,7 @@ func _Admin_CreateService_Handler(srv interface{}, ctx context.Context, dec func } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/CreateService", + FullMethod: Admin_CreateService_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).CreateService(ctx, req.(*CreateServiceRequest)) @@ -274,7 +312,7 @@ func _Admin_UpdateService_Handler(srv interface{}, ctx context.Context, dec func } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/UpdateService", + FullMethod: Admin_UpdateService_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).UpdateService(ctx, req.(*UpdateServiceRequest)) @@ -292,7 +330,7 @@ func _Admin_DeleteService_Handler(srv interface{}, ctx context.Context, dec func } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/DeleteService", + FullMethod: Admin_DeleteService_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).DeleteService(ctx, req.(*DeleteServiceRequest)) @@ -310,7 +348,7 @@ func _Admin_ListTransactions_Handler(srv interface{}, ctx context.Context, dec f } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/ListTransactions", + FullMethod: Admin_ListTransactions_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).ListTransactions(ctx, req.(*ListTransactionsRequest)) @@ -328,7 +366,7 @@ func _Admin_ListTokens_Handler(srv interface{}, ctx context.Context, dec func(in } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/ListTokens", + FullMethod: Admin_ListTokens_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).ListTokens(ctx, req.(*ListTokensRequest)) @@ -346,7 +384,7 @@ func _Admin_RevokeToken_Handler(srv interface{}, ctx context.Context, dec func(i } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/RevokeToken", + FullMethod: Admin_RevokeToken_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).RevokeToken(ctx, req.(*RevokeTokenRequest)) @@ -364,7 +402,7 @@ func _Admin_GetStats_Handler(srv interface{}, ctx context.Context, dec func(inte } info := &grpc.UnaryServerInfo{ Server: srv, - FullMethod: "/adminrpc.Admin/GetStats", + FullMethod: Admin_GetStats_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(AdminServer).GetStats(ctx, req.(*GetStatsRequest)) diff --git a/aperture.go b/aperture.go index 99a760a2..a384cd1b 100644 --- a/aperture.go +++ b/aperture.go @@ -198,6 +198,7 @@ type Aperture struct { httpsServer *http.Server torHTTPServer *http.Server proxy *proxy.Proxy + limiter *staticServiceLimiter proxyCleanup func() adminCleanup func() @@ -396,6 +397,7 @@ func (a *Aperture) Start(errChan chan error, shutdown <-chan struct{}) error { a.cfg, txnStore, secretStore, svcStore, svcHolder.get, func(s []*proxy.Service) error { + a.limiter.refresh(s) if err := a.UpdateServices(s); err != nil { return err } @@ -439,7 +441,7 @@ func (a *Aperture) Start(errChan chan error, shutdown <-chan struct{}) error { txnRecorder = txnStore } - a.proxy, a.proxyCleanup, err = createProxy( + a.proxy, a.limiter, a.proxyCleanup, err = createProxy( a.cfg, initialServices, a.challenger, secretStore, mppSessionStore, paymentSender, mintTxnStore, txnRecorder, adminPriority, adminFallback, @@ -1540,6 +1542,7 @@ func mergeServicesFromDB(configServices []*proxy.Service, Price: row.Price, Auth: auth.Level(row.Auth), AuthScheme: row.AuthScheme, + Timeout: row.Timeout, } } @@ -1633,12 +1636,13 @@ func createProxy(cfg *Config, services []*proxy.Service, paymentSender auth.PaymentSender, txnStore mint.TransactionStore, txnRecorder auth.TransactionRecorder, adminPriorityServices, adminFallbackServices []proxy.LocalService, -) (*proxy.Proxy, func(), error) { +) (*proxy.Proxy, *staticServiceLimiter, func(), error) { + limiter := newStaticServiceLimiter(services) minter := mint.New(&mint.Config{ Challenger: challenger, Secrets: store, - ServiceLimiter: newStaticServiceLimiter(services), + ServiceLimiter: limiter, Now: time.Now, TransactionStore: txnStore, }) @@ -1656,7 +1660,7 @@ func createProxy(cfg *Config, services []*proxy.Service, // it from a deterministic key stored via the secret store. hmacSecret, err := deriveHMACSecret(store) if err != nil { - return nil, nil, fmt.Errorf("MPP HMAC secret: %w", + return nil, nil, nil, fmt.Errorf("MPP HMAC secret: %w", err) } @@ -1716,7 +1720,7 @@ func createProxy(cfg *Config, services []*proxy.Service, staticServer := http.NotFoundHandler() if cfg.ServeStatic { if len(strings.TrimSpace(cfg.StaticRoot)) == 0 { - return nil, nil, fmt.Errorf("staticroot cannot be " + + return nil, nil, nil, fmt.Errorf("staticroot cannot be " + "empty, must contain path to directory that " + "contains index.html") } @@ -1731,7 +1735,7 @@ func createProxy(cfg *Config, services []*proxy.Service, if cfg.HashMail.Enabled { hashMailServices, cleanup, err := createHashMailServer(cfg) if err != nil { - return nil, nil, err + return nil, nil, nil, err } localServices = append(localServices, hashMailServices...) @@ -1755,7 +1759,7 @@ func createProxy(cfg *Config, services []*proxy.Service, authenticator, services, cfg.Blocklist, adminPriorityServices, localServices..., ) - return prxy, proxyCleanup, err + return prxy, limiter, proxyCleanup, err } // createHashMailServer creates the gRPC server for the hash mail message diff --git a/aperturedb/services.go b/aperturedb/services.go index 6ac00b1a..c3c85b2e 100644 --- a/aperturedb/services.go +++ b/aperturedb/services.go @@ -74,6 +74,7 @@ type ServiceParams struct { Auth string AuthScheme string Price int64 + Timeout int64 } // UpsertService inserts or updates a service configuration. @@ -92,6 +93,7 @@ func (s *ServicesStore) UpsertService(ctx context.Context, Price: params.Price, Auth: params.Auth, AuthScheme: params.AuthScheme, + Timeout: params.Timeout, CreatedAt: now, UpdatedAt: now, }) diff --git a/aperturedb/services_test.go b/aperturedb/services_test.go index 98e2b8ea..80794110 100644 --- a/aperturedb/services_test.go +++ b/aperturedb/services_test.go @@ -121,6 +121,70 @@ func TestDeleteService(t *testing.T) { require.Equal(t, "l402+mpp", svcs[0].AuthScheme) } +func TestUpsertServiceTimeout(t *testing.T) { + db := NewTestDB(t) + store := newServicesStoreWithDB(db.BaseDB) + + ctxt, cancel := context.WithTimeout( + context.Background(), defaultTestTimeout, + ) + defer cancel() + + // Insert a service with a non-zero timeout. + err := store.UpsertService(ctxt, ServiceParams{ + Name: "svc-timeout", + Address: "localhost:8080", + Protocol: "http", + HostRegexp: ".*", + PathRegexp: "/api/.*", + AuthScheme: "l402", + Price: 100, + Timeout: 60, + }) + require.NoError(t, err) + + svcs, err := store.ListServices(ctxt) + require.NoError(t, err) + require.Len(t, svcs, 1) + require.Equal(t, int64(60), svcs[0].Timeout) + + // Update to a different timeout value. + err = store.UpsertService(ctxt, ServiceParams{ + Name: "svc-timeout", + Address: "localhost:8080", + Protocol: "http", + HostRegexp: ".*", + PathRegexp: "/api/.*", + AuthScheme: "l402", + Price: 100, + Timeout: 120, + }) + require.NoError(t, err) + + svcs, err = store.ListServices(ctxt) + require.NoError(t, err) + require.Len(t, svcs, 1) + require.Equal(t, int64(120), svcs[0].Timeout) + + // Reset to zero (no expiry). + err = store.UpsertService(ctxt, ServiceParams{ + Name: "svc-timeout", + Address: "localhost:8080", + Protocol: "http", + HostRegexp: ".*", + PathRegexp: "/api/.*", + AuthScheme: "l402", + Price: 100, + Timeout: 0, + }) + require.NoError(t, err) + + svcs, err = store.ListServices(ctxt) + require.NoError(t, err) + require.Len(t, svcs, 1) + require.Equal(t, int64(0), svcs[0].Timeout) +} + func TestListFilteredTransactions(t *testing.T) { db := NewTestDB(t) store := newL402TransactionsStoreWithDB(db.BaseDB) diff --git a/aperturedb/sqlc/migrations/000008_services_timeout.down.sql b/aperturedb/sqlc/migrations/000008_services_timeout.down.sql new file mode 100644 index 00000000..57a1b894 --- /dev/null +++ b/aperturedb/sqlc/migrations/000008_services_timeout.down.sql @@ -0,0 +1 @@ +ALTER TABLE services DROP COLUMN timeout; diff --git a/aperturedb/sqlc/migrations/000008_services_timeout.up.sql b/aperturedb/sqlc/migrations/000008_services_timeout.up.sql new file mode 100644 index 00000000..29591d7f --- /dev/null +++ b/aperturedb/sqlc/migrations/000008_services_timeout.up.sql @@ -0,0 +1 @@ +ALTER TABLE services ADD COLUMN timeout BIGINT NOT NULL DEFAULT 0; diff --git a/aperturedb/sqlc/models.go b/aperturedb/sqlc/models.go index 799d064d..364f2c8f 100644 --- a/aperturedb/sqlc/models.go +++ b/aperturedb/sqlc/models.go @@ -70,4 +70,5 @@ type Service struct { CreatedAt time.Time UpdatedAt time.Time AuthScheme string + Timeout int64 } diff --git a/aperturedb/sqlc/queries/services.sql b/aperturedb/sqlc/queries/services.sql index 9c4bba8a..bf454c8c 100644 --- a/aperturedb/sqlc/queries/services.sql +++ b/aperturedb/sqlc/queries/services.sql @@ -1,9 +1,9 @@ -- name: UpsertService :exec INSERT INTO services ( name, address, protocol, host_regexp, path_regexp, price, auth, - auth_scheme, created_at, updated_at + auth_scheme, timeout, created_at, updated_at ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 ) ON CONFLICT(name) DO UPDATE SET address = excluded.address, @@ -13,6 +13,7 @@ ON CONFLICT(name) DO UPDATE SET price = excluded.price, auth = excluded.auth, auth_scheme = excluded.auth_scheme, + timeout = excluded.timeout, updated_at = excluded.updated_at; -- name: DeleteService :execrows diff --git a/aperturedb/sqlc/services.sql.go b/aperturedb/sqlc/services.sql.go index 80921cdf..d1fe31af 100644 --- a/aperturedb/sqlc/services.sql.go +++ b/aperturedb/sqlc/services.sql.go @@ -24,7 +24,7 @@ func (q *Queries) DeleteService(ctx context.Context, name string) (int64, error) } const listServices = `-- name: ListServices :many -SELECT id, name, address, protocol, host_regexp, path_regexp, price, auth, created_at, updated_at, auth_scheme +SELECT id, name, address, protocol, host_regexp, path_regexp, price, auth, created_at, updated_at, auth_scheme, timeout FROM services ORDER BY name ` @@ -50,6 +50,7 @@ func (q *Queries) ListServices(ctx context.Context) ([]Service, error) { &i.CreatedAt, &i.UpdatedAt, &i.AuthScheme, + &i.Timeout, ); err != nil { return nil, err } @@ -67,9 +68,9 @@ func (q *Queries) ListServices(ctx context.Context) ([]Service, error) { const upsertService = `-- name: UpsertService :exec INSERT INTO services ( name, address, protocol, host_regexp, path_regexp, price, auth, - auth_scheme, created_at, updated_at + auth_scheme, timeout, created_at, updated_at ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 ) ON CONFLICT(name) DO UPDATE SET address = excluded.address, @@ -79,6 +80,7 @@ ON CONFLICT(name) DO UPDATE SET price = excluded.price, auth = excluded.auth, auth_scheme = excluded.auth_scheme, + timeout = excluded.timeout, updated_at = excluded.updated_at ` @@ -91,6 +93,7 @@ type UpsertServiceParams struct { Price int64 Auth string AuthScheme string + Timeout int64 CreatedAt time.Time UpdatedAt time.Time } @@ -105,6 +108,7 @@ func (q *Queries) UpsertService(ctx context.Context, arg UpsertServiceParams) er arg.Price, arg.Auth, arg.AuthScheme, + arg.Timeout, arg.CreatedAt, arg.UpdatedAt, ) diff --git a/docs/admin-api.md b/docs/admin-api.md index f315252c..11a6659b 100644 --- a/docs/admin-api.md +++ b/docs/admin-api.md @@ -88,6 +88,7 @@ curl -X POST \ | `price` | No | 0 | Price in satoshis per request | | `auth` | No | `""` | Auth level: `on`, `off`, or `freebie N` (N free requests per IP) | | `auth_scheme` | No | `AUTH_SCHEME_L402` | Payment auth scheme: `AUTH_SCHEME_L402` (0), `AUTH_SCHEME_MPP` (1), or `AUTH_SCHEME_L402_MPP` (2) | +| `timeout` | No | 0 | Per-service TTL in seconds. When non-zero, minted macaroons include a `_valid_until` caveat so access tokens expire after this many seconds. `0` means no expiry. | ### Update a Service diff --git a/services.go b/services.go index 787bd4ed..a769d7e3 100644 --- a/services.go +++ b/services.go @@ -2,6 +2,7 @@ package aperture import ( "context" + "sync" "time" "github.com/lightninglabs/aperture/l402" @@ -9,10 +10,10 @@ import ( "github.com/lightninglabs/aperture/proxy" ) -// staticServiceLimiter provides static restrictions for services. -// -// TODO(wilmer): use etcd instead. +// staticServiceLimiter provides live-updatable restrictions for services. Its +// maps are rebuilt atomically via refresh whenever the service list changes. type staticServiceLimiter struct { + mu sync.RWMutex capabilities map[l402.Service]l402.Caveat constraints map[l402.Service][]l402.Caveat timeouts map[l402.Service]int64 @@ -27,9 +28,17 @@ var _ mint.ServiceLimiter = (*staticServiceLimiter)(nil) func newStaticServiceLimiter( proxyServices []*proxy.Service) *staticServiceLimiter { - capabilities := make(map[l402.Service]l402.Caveat) - constraints := make(map[l402.Service][]l402.Caveat) - timeouts := make(map[l402.Service]int64) + l := &staticServiceLimiter{} + l.refresh(proxyServices) + return l +} + +// refresh rebuilds all three service maps atomically from the given service +// list. It is safe to call concurrently with the Service* read methods. +func (l *staticServiceLimiter) refresh(proxyServices []*proxy.Service) { + caps := make(map[l402.Service]l402.Caveat) + cons := make(map[l402.Service][]l402.Caveat) + tos := make(map[l402.Service]int64) for _, proxyService := range proxyServices { s := l402.Service{ @@ -39,23 +48,23 @@ func newStaticServiceLimiter( } if proxyService.Timeout > 0 { - timeouts[s] = proxyService.Timeout + tos[s] = proxyService.Timeout } - capabilities[s] = l402.NewCapabilitiesCaveat( + caps[s] = l402.NewCapabilitiesCaveat( proxyService.Name, proxyService.Capabilities, ) for cond, value := range proxyService.Constraints { caveat := l402.Caveat{Condition: cond, Value: value} - constraints[s] = append(constraints[s], caveat) + cons[s] = append(cons[s], caveat) } } - return &staticServiceLimiter{ - capabilities: capabilities, - constraints: constraints, - timeouts: timeouts, - } + l.mu.Lock() + l.capabilities = caps + l.constraints = cons + l.timeouts = tos + l.mu.Unlock() } // ServiceCapabilities returns the capabilities caveats for each service. This @@ -63,6 +72,9 @@ func newStaticServiceLimiter( func (l *staticServiceLimiter) ServiceCapabilities(ctx context.Context, services ...l402.Service) ([]l402.Caveat, error) { + l.mu.RLock() + defer l.mu.RUnlock() + res := make([]l402.Caveat, 0, len(services)) for _, service := range services { capabilities, ok := l.capabilities[service] @@ -80,6 +92,9 @@ func (l *staticServiceLimiter) ServiceCapabilities(ctx context.Context, func (l *staticServiceLimiter) ServiceConstraints(ctx context.Context, services ...l402.Service) ([]l402.Caveat, error) { + l.mu.RLock() + defer l.mu.RUnlock() + res := make([]l402.Caveat, 0, len(services)) for _, service := range services { constraints, ok := l.constraints[service] @@ -97,6 +112,9 @@ func (l *staticServiceLimiter) ServiceConstraints(ctx context.Context, func (l *staticServiceLimiter) ServiceTimeouts(ctx context.Context, services ...l402.Service) ([]l402.Caveat, error) { + l.mu.RLock() + defer l.mu.RUnlock() + res := make([]l402.Caveat, 0, len(services)) for _, service := range services { numSeconds, ok := l.timeouts[service] diff --git a/services_test.go b/services_test.go index 0dcee9a1..6425e9f2 100644 --- a/services_test.go +++ b/services_test.go @@ -2,6 +2,7 @@ package aperture import ( "context" + "sync" "testing" "time" @@ -54,6 +55,109 @@ func TestServiceTimeoutsComputedPerCall(t *testing.T) { "from init time") } +// TestRefreshRebuildsTimeouts verifies that calling refresh replaces the +// timeout map so subsequent ServiceTimeouts calls reflect the new values. +func TestRefreshRebuildsTimeouts(t *testing.T) { + t.Parallel() + + limiter := newStaticServiceLimiter([]*proxy.Service{ + {Name: "svc", Price: 10, Timeout: 60}, + }) + + svc := l402.Service{Name: "svc", Tier: l402.BaseTier, Price: 10} + + caveats, err := limiter.ServiceTimeouts(context.Background(), svc) + require.NoError(t, err) + require.Len(t, caveats, 1) + + // Refresh with an updated timeout. + limiter.refresh([]*proxy.Service{ + {Name: "svc", Price: 10, Timeout: 120}, + }) + + caveats2, err := limiter.ServiceTimeouts(context.Background(), svc) + require.NoError(t, err) + require.Len(t, caveats2, 1) + require.NotEqual(t, caveats[0].Value, caveats2[0].Value) + + // Refresh removing the timeout entirely (Timeout == 0). + limiter.refresh([]*proxy.Service{ + {Name: "svc", Price: 10, Timeout: 0}, + }) + + caveats3, err := limiter.ServiceTimeouts(context.Background(), svc) + require.NoError(t, err) + require.Empty(t, caveats3) +} + +// TestRefreshConcurrentReads verifies that concurrent reads during a refresh +// do not race. Run with -race to catch data races. +func TestRefreshConcurrentReads(t *testing.T) { + t.Parallel() + + limiter := newStaticServiceLimiter([]*proxy.Service{ + {Name: "svc", Price: 5, Timeout: 30}, + }) + + svc := l402.Service{Name: "svc", Tier: l402.BaseTier, Price: 5} + + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func(timeout int64) { + defer wg.Done() + limiter.refresh([]*proxy.Service{ + {Name: "svc", Price: 5, Timeout: timeout}, + }) + }(int64(i + 1)) + + wg.Add(1) + go func() { + defer wg.Done() + _, _ = limiter.ServiceTimeouts(context.Background(), svc) + _, _ = limiter.ServiceCapabilities( + context.Background(), svc, + ) + _, _ = limiter.ServiceConstraints( + context.Background(), svc, + ) + }() + } + wg.Wait() +} + +// TestRefreshCreateDelete verifies that refresh propagates creates and deletes: +// a service added via refresh becomes visible, and one removed disappears. +func TestRefreshCreateDelete(t *testing.T) { + t.Parallel() + + limiter := newStaticServiceLimiter([]*proxy.Service{ + {Name: "existing", Price: 1, Timeout: 10}, + }) + + existing := l402.Service{Name: "existing", Tier: l402.BaseTier, Price: 1} + added := l402.Service{Name: "new-svc", Tier: l402.BaseTier, Price: 2} + + // Create: refresh with an additional service. + limiter.refresh([]*proxy.Service{ + {Name: "existing", Price: 1, Timeout: 10}, + {Name: "new-svc", Price: 2, Timeout: 20}, + }) + + caveats, err := limiter.ServiceTimeouts(context.Background(), added) + require.NoError(t, err) + require.Len(t, caveats, 1) + + // Delete: refresh without the original service. + limiter.refresh([]*proxy.Service{ + {Name: "new-svc", Price: 2, Timeout: 20}, + }) + + caveats, err = limiter.ServiceTimeouts(context.Background(), existing) + require.NoError(t, err) + require.Empty(t, caveats) +} + func TestStaticServiceLimiterAllCaveatTypes(t *testing.T) { t.Parallel()