111 lines
3.0 KiB
Go
111 lines
3.0 KiB
Go
package runnerfacade
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
|
|
)
|
|
|
|
type claim struct {
|
|
assignment taskassignment.Assignment
|
|
claimed bool
|
|
ready chan struct{}
|
|
}
|
|
|
|
// Registry holds pending task payloads and an active authorization cache.
|
|
// Pending entries are rebuilt by JetStream redelivery; active authorization is
|
|
// reconstructable from backend metadata and is not an independent state store.
|
|
type Registry struct {
|
|
mu sync.Mutex
|
|
claims map[string]*claim
|
|
}
|
|
|
|
func NewRegistry() *Registry { return &Registry{claims: make(map[string]*claim)} }
|
|
|
|
func (r *Registry) Offer(assignment taskassignment.Assignment) (<-chan struct{}, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if existing := r.claims[assignment.ID]; existing != nil {
|
|
if existing.assignment.Identity != assignment.Identity || existing.assignment.Task.GetId() != assignment.Task.GetId() {
|
|
return nil, fmt.Errorf("assignment %s was offered with different task data", assignment.ID)
|
|
}
|
|
return existing.ready, nil
|
|
}
|
|
entry := &claim{assignment: assignment, ready: make(chan struct{})}
|
|
r.claims[assignment.ID] = entry
|
|
return entry.ready, nil
|
|
}
|
|
|
|
func (r *Registry) Claim(assignmentID, spiffeID string) (taskassignment.Assignment, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
entry := r.claims[assignmentID]
|
|
if entry == nil {
|
|
return taskassignment.Assignment{}, errors.New("assignment is not pending")
|
|
}
|
|
if entry.assignment.Identity.SPIFFEID != spiffeID {
|
|
return taskassignment.Assignment{}, errors.New("executor SPIFFE ID does not match assignment")
|
|
}
|
|
if entry.claimed {
|
|
return taskassignment.Assignment{}, errors.New("assignment was already claimed")
|
|
}
|
|
entry.claimed = true
|
|
close(entry.ready)
|
|
return entry.assignment, nil
|
|
}
|
|
|
|
func (r *Registry) Resolve(assignmentID, spiffeID string) (taskassignment.Assignment, error) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
entry := r.claims[assignmentID]
|
|
if entry == nil || !entry.claimed {
|
|
return taskassignment.Assignment{}, errors.New("assignment is not claimed")
|
|
}
|
|
if entry.assignment.Identity.SPIFFEID != spiffeID {
|
|
return taskassignment.Assignment{}, errors.New("executor SPIFFE ID does not match assignment")
|
|
}
|
|
return entry.assignment, nil
|
|
}
|
|
|
|
func (r *Registry) WaitClaimed(ctx context.Context, assignmentID string) error {
|
|
r.mu.Lock()
|
|
entry := r.claims[assignmentID]
|
|
r.mu.Unlock()
|
|
if entry == nil {
|
|
return errors.New("assignment is not pending")
|
|
}
|
|
select {
|
|
case <-entry.ready:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (r *Registry) Remove(assignmentID string) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
delete(r.claims, assignmentID)
|
|
}
|
|
|
|
// RecoverClaimed restores authorization for an executor that already claimed
|
|
// its task before the controller restarted.
|
|
func (r *Registry) RecoverClaimed(assignment taskassignment.Assignment) error {
|
|
ready, err := r.Offer(assignment)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
r.mu.Lock()
|
|
entry := r.claims[assignment.ID]
|
|
if !entry.claimed {
|
|
entry.claimed = true
|
|
close(entry.ready)
|
|
}
|
|
r.mu.Unlock()
|
|
<-ready
|
|
return nil
|
|
}
|