package requester import ( "context" "crypto/rand" "crypto/subtle" "encoding/hex" "encoding/json" "errors" "fmt" "strings" "time" "github.com/bluesky-social/indigo/atproto/syntax" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" commonv1 "github.com/johnandersen777/socialweb-computer-kcp/api/common/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/cloudinit" "github.com/johnandersen777/socialweb-computer-kcp/internal/atproto/market" "github.com/johnandersen777/socialweb-computer-kcp/internal/policyengine/trust" ) const ( ContractAssociationLifetime = market.ContractAssociationLifetime associationRetryFirst = 2 * time.Second associationRetryMax = 5 * time.Minute associationRetryAttempts = 12 associationNonceBytes = 24 associationStepTimeout = 30 * time.Second ) var errAssociationRefused = errors.New("requester: the association is refused") type contractAssociation struct { subject string rkey string ref commonv1.StrongRef expiresAt time.Time finished bool } type associationRequestEntry struct { Authority string `json:"authority"` Service string `json:"service"` Contract string `json:"contract"` Nonce string `json:"nonce"` } func associationSubject(contract v1alpha1.ComputeContract, state *v1alpha1.NodeClaimContract) string { a := contract.Spec.Association if a == nil { return "" } subject := strings.TrimSpace(a.Subject) switch { case a.FromGuest && subject != "": return "" case subject != "": return subject case a.FromGuest && state != nil && state.GuestRequester != nil && state.GuestRequester.State == v1alpha1.ContractGuestRequesterVerified: return strings.TrimSpace(state.GuestRequester.Did) } return "" } func fromGuest(contract v1alpha1.ComputeContract) bool { a := contract.Spec.Association return a != nil && a.FromGuest && strings.TrimSpace(a.Subject) == "" } func associationRefFile(contract v1alpha1.ComputeContract) (string, error) { if contract.Spec.Association == nil { return cloudinit.DefaultAssociationRefFile, nil } return cloudinit.AssociationRefPath(contract.Spec.Association.RefFile) } func ContractAssociationRKey(subject string, accept commonv1.StrongRef) string { return market.ContractScopedRKey(trust.RequesterAssociate, subject, accept.URI) } func (r *contractRunner) associationWritten(subject string) bool { r.mu.Lock() defer r.mu.Unlock() return r.association != nil && r.association.subject == subject } func (r *contractRunner) associationFinished(subject string) bool { r.mu.Lock() defer r.mu.Unlock() return r.association != nil && r.association.subject == subject && r.association.finished } func guestPending(state *v1alpha1.NodeClaimContract) bool { return state != nil && state.GuestRequester != nil && state.GuestRequester.State == v1alpha1.ContractGuestRequesterPending } func (r *contractRunner) associationPending(contract v1alpha1.ComputeContract) bool { state := contract.Status.Contract if state == nil || state.Accept == nil || state.Accept.URI == "" || contract.Spec.Release || contract.DeletionTimestamp != nil { return false } if fromGuest(contract) && guestPending(state) { return true } subject := associationSubject(contract, state) return subject != "" && !r.associationFinished(subject) } func (r *contractRunner) associateLater(contract v1alpha1.ComputeContract) { if r.associationPending(contract) { r.kickAssociation() } } func (r *contractRunner) kickAssociation() { select { case r.associationKick <- struct{}{}: default: } } func (r *contractRunner) associationStep(ctx context.Context, state *v1alpha1.NodeClaimContract) { if state == nil || r.isGone() { return } stepCtx, cancel := context.WithTimeout(ctx, associationStepTimeout) defer cancel() r.ensureAssociation(stepCtx, state.Accept, state) } func (r *contractRunner) associationFailed(ctx context.Context, step string, err error) bool { refused := errors.Is(err, errAssociationRefused) || errors.Is(err, errAtircLeaseMissing) r.log().Warn("requester_association_failed", "step", step, "err", err.Error(), "retry", !refused) message := step + ": " + err.Error() r.mu.Lock() attempts := r.associationAttempts if !refused { r.associationAttempts++ } r.mu.Unlock() terminal := refused || attempts >= associationRetryAttempts switch { case refused: case terminal: message += fmt.Sprintf(" (gave up after %d attempts)", attempts) default: delay := min(associationRetryFirst<