Remove run node UI and expire registrations
This commit is contained in:
@@ -168,7 +168,7 @@ func TestRunSessionPersistsAndSignedEnvelopeRejectsReplay(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("reload store: %v", err)
|
||||
}
|
||||
reloaded := newCoreService(reloadedStore, func() time.Time { return now.Add(time.Minute) })
|
||||
reloaded := newCoreService(reloadedStore, func() time.Time { return now.Add(runHeartbeatStaleAfter / 2) })
|
||||
result, err := reloaded.AcceptRunHeartbeat(domain.RunControlHeartbeat{
|
||||
RunEndpointID: "run-local", SessionToken: hello.SessionToken, Version: "0.1.1",
|
||||
Status: domain.RunEndpointStatusOnline, CapabilityFingerprint: "cap-jobs",
|
||||
|
||||
@@ -73,6 +73,9 @@ func (svc *CoreService) RegisterRunHello(hello domain.RunControlHello) (domain.R
|
||||
svc.controlMu.Lock()
|
||||
defer svc.controlMu.Unlock()
|
||||
|
||||
if err := svc.sweepExpiredRunRegistrationsLocked(stamp); err != nil {
|
||||
return domain.RunControlHelloResult{}, err
|
||||
}
|
||||
if err := svc.upsertRunEndpoint(endpoint); err != nil {
|
||||
return domain.RunControlHelloResult{}, err
|
||||
}
|
||||
@@ -183,6 +186,9 @@ func (svc *CoreService) AcceptRunHeartbeat(heartbeat domain.RunControlHeartbeat)
|
||||
svc.controlMu.Lock()
|
||||
defer svc.controlMu.Unlock()
|
||||
|
||||
if err := svc.sweepExpiredRunRegistrationsLocked(stamp); err != nil {
|
||||
return domain.RunControlHeartbeatResult{}, err
|
||||
}
|
||||
session, err := svc.currentRunSession(heartbeat.RunEndpointID, heartbeat.SessionToken)
|
||||
if err != nil {
|
||||
return domain.RunControlHeartbeatResult{}, err
|
||||
@@ -230,6 +236,43 @@ func (svc *CoreService) upsertRunEndpoint(endpoint domain.RunEndpoint) error {
|
||||
return svc.store.RunEndpoints().Update(endpoint)
|
||||
}
|
||||
|
||||
func (svc *CoreService) sweepExpiredRunRegistrationsLocked(stamp time.Time) error {
|
||||
endpoints, err := svc.store.RunEndpoints().List(domain.RunEndpointFilter{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, endpoint := range endpoints {
|
||||
if runEndpointRegistrationCurrentAt(endpoint, stamp) {
|
||||
continue
|
||||
}
|
||||
if err := svc.store.RunEndpoints().Delete(endpoint.ID); err != nil && !errors.Is(err, repo.ErrNotFound) {
|
||||
return err
|
||||
}
|
||||
delete(svc.runSessions, endpoint.ID)
|
||||
session, err := svc.store.RunControlSessions().Get(endpoint.ID)
|
||||
if err != nil {
|
||||
if errors.Is(err, repo.ErrNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
}
|
||||
if session.Status != domain.AuthSessionStatusActive {
|
||||
continue
|
||||
}
|
||||
session.Status = domain.AuthSessionStatusRevoked
|
||||
session.RevokedAt = stamp
|
||||
session.UpdatedAt = stamp
|
||||
if err := svc.store.RunControlSessions().Update(session); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func runEndpointRegistrationCurrentAt(endpoint domain.RunEndpoint, stamp time.Time) bool {
|
||||
return !endpoint.LastHeartbeatAt.IsZero() && !stamp.After(endpoint.LastHeartbeatAt.Add(runHeartbeatStaleAfter))
|
||||
}
|
||||
|
||||
func (svc *CoreService) nextSessionToken() (string, error) {
|
||||
return randomToken()
|
||||
}
|
||||
|
||||
@@ -118,6 +118,40 @@ func TestCoreServiceAcceptsRunHeartbeat(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoreServiceClearsRunRegistrationAfterHeartbeatTTL(t *testing.T) {
|
||||
svc := newTestCoreService()
|
||||
hello, err := svc.RegisterRunHello(validRunControlHello())
|
||||
if err != nil {
|
||||
t.Fatalf("register hello: %v", err)
|
||||
}
|
||||
|
||||
svc.now = func() time.Time { return fixedTime.Add(runHeartbeatStaleAfter - time.Nanosecond) }
|
||||
active, err := svc.ListRunEndpoints(domain.RunEndpointFilter{Status: domain.RunEndpointStatusOnline})
|
||||
if err != nil || len(active) != 1 || active[0].ID != "run-local" {
|
||||
t.Fatalf("expected registration inside TTL, endpoints=%+v err=%v", active, err)
|
||||
}
|
||||
|
||||
svc.now = func() time.Time { return fixedTime.Add(runHeartbeatStaleAfter + time.Second) }
|
||||
if _, err := svc.GetRunEndpoint("run-local"); !errors.Is(err, repo.ErrNotFound) {
|
||||
t.Fatalf("expected expired registration to be cleared, got %v", err)
|
||||
}
|
||||
cleared, err := svc.ListRunEndpoints(domain.RunEndpointFilter{})
|
||||
if err != nil || len(cleared) != 0 {
|
||||
t.Fatalf("expected no stale registrations in list, endpoints=%+v err=%v", cleared, err)
|
||||
}
|
||||
_, err = svc.AcceptRunHeartbeat(domain.RunControlHeartbeat{
|
||||
RunEndpointID: "run-local",
|
||||
SessionToken: hello.SessionToken,
|
||||
Version: "0.1.1",
|
||||
Status: domain.RunEndpointStatusOnline,
|
||||
CapabilityFingerprint: "cap-v1",
|
||||
Capacity: domain.RunCapacity{MaxJobs: 4},
|
||||
})
|
||||
if err == nil || !strings.Contains(err.Error(), "sessionToken is invalid") {
|
||||
t.Fatalf("expected stale session to be rejected after registration cleanup, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoreServiceRejectsInvalidRunHeartbeatToken(t *testing.T) {
|
||||
svc := newTestCoreService()
|
||||
if _, err := svc.RegisterRunHello(validRunControlHello()); err != nil {
|
||||
|
||||
@@ -25,7 +25,7 @@ var (
|
||||
|
||||
const (
|
||||
ServerDeletionForceConfirmation = "FORCE DELETE"
|
||||
runHeartbeatStaleAfter = 2 * time.Minute
|
||||
runHeartbeatStaleAfter = 30 * time.Second
|
||||
)
|
||||
|
||||
type ForbiddenError struct {
|
||||
@@ -1609,10 +1609,20 @@ func (svc *CoreService) CreateRunEndpoint(endpoint domain.RunEndpoint) (domain.R
|
||||
}
|
||||
|
||||
func (svc *CoreService) GetRunEndpoint(id string) (domain.RunEndpoint, error) {
|
||||
svc.controlMu.Lock()
|
||||
defer svc.controlMu.Unlock()
|
||||
if err := svc.sweepExpiredRunRegistrationsLocked(svc.now()); err != nil {
|
||||
return domain.RunEndpoint{}, err
|
||||
}
|
||||
return svc.store.RunEndpoints().Get(id)
|
||||
}
|
||||
|
||||
func (svc *CoreService) ListRunEndpoints(filter domain.RunEndpointFilter) ([]domain.RunEndpoint, error) {
|
||||
svc.controlMu.Lock()
|
||||
defer svc.controlMu.Unlock()
|
||||
if err := svc.sweepExpiredRunRegistrationsLocked(svc.now()); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return svc.store.RunEndpoints().List(filter)
|
||||
}
|
||||
|
||||
@@ -2645,10 +2655,7 @@ func (svc *CoreService) validateRunnableEndpoint(endpoint domain.RunEndpoint, ca
|
||||
}
|
||||
|
||||
func (svc *CoreService) runEndpointHeartbeatCurrent(endpoint domain.RunEndpoint) bool {
|
||||
if endpoint.LastHeartbeatAt.IsZero() {
|
||||
return false
|
||||
}
|
||||
return !svc.now().After(endpoint.LastHeartbeatAt.Add(runHeartbeatStaleAfter))
|
||||
return runEndpointRegistrationCurrentAt(endpoint, svc.now())
|
||||
}
|
||||
|
||||
func maxInt(a, b int) int {
|
||||
|
||||
Reference in New Issue
Block a user