package requester import ( "cmp" "context" "crypto/sha256" "encoding/hex" "errors" "fmt" "strconv" "strings" "time" apiequality "k8s.io/apimachinery/pkg/api/equality" 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/market" contractmarket "github.com/johnandersen777/socialweb-computer-kcp/internal/contract/market" "github.com/johnandersen777/socialweb-computer-kcp/internal/kit/conds" "github.com/johnandersen777/socialweb-computer-kcp/internal/market/participant" ) const ( ContractEventSubmitRetry = 2 * time.Second ContractEventSubmitRetryCap = time.Minute ContractEventSubmitTimeout = 10 * time.Second inboundContractEventPrefix = "in-" maxGuestKey = 256 maxGuestCiphertext = 16384 ageArmorHeader = "-----BEGIN AGE ENCRYPTED FILE-----" ) var _ contractmarket.ContractEventObserver = (*Requester)(nil) type guestPayload struct { Kind string Tag string Key string Ciphertext string CreatedAt string } func InboundContractEventName(eventURI string) string { sum := sha256.Sum256([]byte(eventURI)) return inboundContractEventPrefix + hex.EncodeToString(sum[:10]) } func validateGuestFields(nsid, kind, tag, key, ciphertext string) error { if nsid == market.ComputeEventsGuestSecretsNSID { switch { case kind != "" || tag != "" || key != "": return errors.New("a secrets event carries only a ciphertext") case len(ciphertext) > market.MaxGuestSecretsCiphertext || !strings.HasPrefix(strings.TrimSpace(ciphertext), ageArmorHeader): return fmt.Errorf("ciphertext is not age armored ciphertext of at most %d bytes", market.MaxGuestSecretsCiphertext) } return nil } if nsid == market.ComputeEventsGuestAgentStartNSID || nsid == market.ComputeEventsGuestSecretsUpdatedNSID || nsid == market.ComputeEventsGuestR2StartNSID || nsid == market.ComputeEventsGuestR2ReleaseNSID { if kind != "" || tag != "" || key != "" || ciphertext != "" { return errors.New("an agentStart or secretsUpdated event carries no kind, tag, key or ciphertext") } return nil } if !networkTagRE.MatchString(tag) { return fmt.Errorf("tag %q is not a DNS label of at most 32 characters", tag) } switch nsid { case market.ComputeEventsGuestPublicKeyNSID: switch { case kind != market.GuestPublicKeyKindAgeX25519: return fmt.Errorf("kind %q is not %s", kind, market.GuestPublicKeyKindAgeX25519) case key == "" || len(key) > maxGuestKey || !strings.HasPrefix(key, "age1") || strings.ContainsAny(key, " \t\r\n"): return fmt.Errorf("key is not an age X25519 recipient (age1..., at most %d bytes)", maxGuestKey) case ciphertext != "": return errors.New("a publicKey carries no ciphertext") } case market.ComputeEventsGuestSignalWebRTCNSID: switch { case kind != market.GuestSignalKindOffer && kind != market.GuestSignalKindAnswer: return fmt.Errorf("kind %q is not %s or %s", kind, market.GuestSignalKindOffer, market.GuestSignalKindAnswer) case len(ciphertext) > maxGuestCiphertext || !strings.HasPrefix(strings.TrimSpace(ciphertext), ageArmorHeader): return fmt.Errorf("ciphertext is not age armored ciphertext of at most %d bytes", maxGuestCiphertext) case key != "": return errors.New("a signalWebRTC carries no key") } default: return fmt.Errorf("nsid %q is not a guest event", nsid) } return nil } func parseGuestPayload(nsid string, value map[string]any) (guestPayload, error) { var out guestPayload switch nsid { case market.ComputeEventsGuestPublicKeyNSID: record, err := market.ParseComputeEventsGuestPublicKey(value) if err != nil { return out, err } out = guestPayload{Kind: record.Kind, Tag: record.Tag, Key: record.Key, CreatedAt: record.CreatedAt} case market.ComputeEventsGuestSignalWebRTCNSID: record, err := market.ParseComputeEventsGuestSignalWebRTC(value) if err != nil { return out, err } out = guestPayload{Kind: record.Kind, Tag: record.Tag, Ciphertext: record.Ciphertext, CreatedAt: record.CreatedAt} default: return out, fmt.Errorf("nsid %q is not a guest event", nsid) } if err := validateGuestFields(nsid, out.Kind, out.Tag, out.Key, out.Ciphertext); err != nil { return out, err } if _, err := market.ParseTime(out.CreatedAt); err != nil { return out, fmt.Errorf("createdAt: %w", err) } return out, nil } func outboundKind(spec v1alpha1.ContractEventSpec) string { if spec.NSID == market.ComputeEventsGuestPublicKeyNSID { return cmp.Or(spec.Kind, market.GuestPublicKeyKindAgeX25519) } return spec.Kind } func (q *Requester) ObserveContractEvent(ctx context.Context, change contractmarket.ContractEventChange) (time.Duration, error) { if change.Gone || change.Event == nil { q.contractEvents.Remove(change.Name) return 0, nil } event := *change.Event if contract := change.Contract; contract != nil && !q.ownsContract(*contract) { return 0, nil } if prev, ok := q.contractEvents.Get(event.Name); ok && (prev.UID == "" || prev.UID == event.UID) && statusAhead(prev.Status, event.Status) { event.Status = *prev.Status.DeepCopy() } q.contractEvents.Put(event.Name, event) if q.stopping() || q.deps.ContractEvents == nil || event.DeletionTimestamp != nil { return 0, nil } if contract := change.Contract; contract != nil && contract.UID != "" && !ownedBy(event, *contract) { if err := q.deps.ContractEvents.ApplyContractEventOwner(ctx, event.Name, *contract); err != nil { return 0, err } event.OwnerReferences = append(event.OwnerReferences, contractOwnerReference(*contract)) q.contractEvents.Put(event.Name, event) } if event.Spec.Direction == v1alpha1.ContractEventInbound { if err := q.observeInboundContractEvent(ctx, event); err != nil { return 0, err } if contract := change.Contract; contract != nil && guestSecretsKey(event) { return 0, q.sealGuestSecrets(ctx, *contract) } return 0, nil } return q.observeOutboundContractEvent(ctx, event, change.Contract) } func contractEventRank(status v1alpha1.ContractEventStatus) int { switch status.Phase { case v1alpha1.ContractEventSubmitted, v1alpha1.ContractEventReceived, v1alpha1.ContractEventFailed: return 3 case v1alpha1.ContractEventPublished: return 2 } if status.Record != nil { return 2 } return 1 } func submitAttempts(status v1alpha1.ContractEventStatus) int32 { if status.Submit == nil { return 0 } return status.Submit.Attempts } func statusAhead(known, seen v1alpha1.ContractEventStatus) bool { if known.ObservedGeneration != seen.ObservedGeneration { return known.ObservedGeneration > seen.ObservedGeneration } if a, b := contractEventRank(known), contractEventRank(seen); a != b { return a > b } return submitAttempts(known) > submitAttempts(seen) } func (q *Requester) writeContractEventStatus(ctx context.Context, event *v1alpha1.ContractEvent, status v1alpha1.ContractEventStatus) error { status.ObservedGeneration = event.Generation if apiequality.Semantic.DeepEqual(event.Status, status) { return nil } if err := q.deps.ContractEvents.ApplyContractEventStatus(ctx, event.Name, status); err != nil { return err } event.Status = *status.DeepCopy() q.contractEvents.Put(event.Name, *event) return nil } func (q *Requester) observeInboundContractEvent(ctx context.Context, event v1alpha1.ContractEvent) error { if event.Status.Record != nil { return nil } if _, creating := q.creatingInbound.Load(event.Name); creating || event.Status.Phase == v1alpha1.ContractEventFailed { return nil } status := *event.Status.DeepCopy() status.Phase = v1alpha1.ContractEventFailed status.Message = "an inbound ContractEvent is written by the requester from a bidder event; create an outbound one" q.log.Warn("requester_contract_event_refused", "contractEvent", event.Name, "reason", status.Message) return q.writeContractEventStatus(ctx, &event, status) } func (q *Requester) observeOutboundContractEvent(ctx context.Context, event v1alpha1.ContractEvent, contract *v1alpha1.ComputeContract) (time.Duration, error) { respec := event.Status.ObservedGeneration != 0 && event.Status.ObservedGeneration != event.Generation if !respec && (event.Status.Phase == v1alpha1.ContractEventFailed || event.Status.Phase == v1alpha1.ContractEventSubmitted) { return 0, nil } now := q.now() status := *event.Status.DeepCopy() if respec { status = v1alpha1.ContractEventStatus{Conditions: status.Conditions} } fail := func(message string) (time.Duration, error) { status.Phase = v1alpha1.ContractEventFailed status.Message = message if status.Record == nil { conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionPublished, metav1.ConditionFalse, "Failed", message, now) } conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionSubmitted, metav1.ConditionFalse, "Failed", message, now) q.log.Warn("requester_contract_event_failed", "contractEvent", event.Name, "contract", event.Spec.ContractRef.Name, "reason", message) return 0, q.writeContractEventStatus(ctx, &event, status) } wait := func(message string) (time.Duration, error) { status.Phase = v1alpha1.ContractEventPending status.Message = message conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionPublished, metav1.ConditionFalse, "Waiting", message, now) return 0, q.writeContractEventStatus(ctx, &event, status) } spec := event.Spec if err := validateGuestFields(spec.NSID, outboundKind(spec), spec.Tag, spec.Key, spec.Ciphertext); err != nil { return fail("the spec is invalid: " + err.Error()) } name := strings.TrimSpace(spec.ContractRef.Name) if contract == nil { if status.Record != nil { return fail("the ComputeContract " + name + " is gone") } return wait("waiting for the ComputeContract " + name) } if contract.DeletionTimestamp != nil { return fail("the ComputeContract " + name + " is being deleted") } if secretsEvent(spec.NSID) && q.releasing(*contract) { if status.Record != nil { return 0, nil } return wait("the release of the ComputeContract " + name + " is requested; its secrets are not published") } if spec.NSID == market.ComputeEventsGuestAgentStartNSID && contract.Spec.Agent == nil { return fail("the ComputeContract " + name + " has no spec.agent; an agentStart starts nothing") } if spec.NSID == market.ComputeEventsGuestSecretsUpdatedNSID && contract.Spec.SecretsRef == nil { return fail("the ComputeContract " + name + " has no spec.secretsRef; a secretsUpdated updates nothing") } if (spec.NSID == market.ComputeEventsGuestR2StartNSID || spec.NSID == market.ComputeEventsGuestR2ReleaseNSID) && !fromGuest(*contract) { return fail("the ComputeContract " + name + " has no spec.association.fromGuest; it runs no guest requester") } if spec.NSID == market.ComputeEventsGuestR2StartNSID { if spec.R2Start == nil { return fail("an r2Start event names no spec.r2Start") } if err := market.ValidGuestRequesterName(spec.R2Start.Name); err != nil { return fail(err.Error()) } } state := contract.Status.Contract switch contract.Status.Phase { case v1alpha1.ComputeContractReleased, v1alpha1.ComputeContractFailed: return fail("the ComputeContract " + name + " is " + string(contract.Status.Phase)) case v1alpha1.ComputeContractHeld, v1alpha1.ComputeContractReleasing: default: if status.Record == nil { return wait("waiting for the ComputeContract " + name + " to be held") } } if state == nil || state.Receipt == nil || state.Receipt.URI == "" || state.Accept == nil || state.Accept.URI == "" { return wait("waiting for the receipt and the accept of the ComputeContract " + name) } envelope, err := q.publishContractEvent(ctx, event, state) if err != nil { return 0, fmt.Errorf("requester: publishing the ContractEvent %s: %w", event.Name, err) } if status.Record == nil || status.Event == nil || status.Phase != v1alpha1.ContractEventPublished { status.Record = &commonv1.StrongRef{URI: envelope.payload.URI, CID: envelope.payload.CID} status.Event = &commonv1.StrongRef{URI: envelope.event.URI, CID: envelope.event.CID} status.Phase = v1alpha1.ContractEventPublished status.Message = "the record and its market.event are in the requester's repo" conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionPublished, metav1.ConditionTrue, "Published", status.Message, now) if err := q.writeContractEventStatus(ctx, &event, status); err != nil { return 0, err } q.log.Info("requester_contract_event_published", "contractEvent", event.Name, "nsid", spec.NSID, "tag", spec.Tag, "record", status.Record.URI, "event", status.Event.URI) } return q.submitContractEvent(ctx, &event, status, state, envelope.event) } func ownedBy(event v1alpha1.ContractEvent, contract v1alpha1.ComputeContract) bool { for _, ref := range event.OwnerReferences { if ref.Kind == "ComputeContract" && ref.Name == contract.Name && ref.UID == contract.UID { return true } } return false } func contractOwnerReference(contract v1alpha1.ComputeContract) metav1.OwnerReference { return metav1.OwnerReference{APIVersion: v1alpha1.GroupVersion.String(), Kind: "ComputeContract", Name: contract.Name, UID: contract.UID} } type publishedContractEvent struct { payload market.SignedRecord event market.SignedRecord } func (q *Requester) publishContractEvent(ctx context.Context, event v1alpha1.ContractEvent, state *v1alpha1.NodeClaimContract) (publishedContractEvent, error) { var out publishedContractEvent uid := string(event.UID) if uid == "" { return out, errors.New("the ContractEvent has no UID") } nsid := event.Spec.NSID if event.Generation > 1 { uid += "-g" + strconv.FormatInt(event.Generation, 10) } payloadRKey := sessionRKey("contract-event", uid) payload, found, err := q.existingSignedRecord(ctx, nsid, payloadRKey) if err != nil { return out, fmt.Errorf("read the payload record: %w", err) } if !found { record, err := q.contractEventRecord(event) if err != nil { return out, err } payload, err = market.CreateSignedRecordAt(ctx, q.repo, q.repoDID, nsid, payloadRKey, record, q.signer) if err != nil { return out, fmt.Errorf("create the payload record: %w", err) } } eventRKey := sessionRKey("contract-event-envelope", uid) envelope, found, err := q.existingSignedRecord(ctx, market.EventNSID, eventRKey) if err != nil { return out, fmt.Errorf("read the market.event: %w", err) } if !found { envelope, err = market.PutEvent(ctx, q.repo, q.repoDID, eventRKey, market.EventOptions{ Receipt: market.RecordRef{URI: state.Receipt.URI, CID: state.Receipt.CID}, Payload: market.RecordRef{URI: payload.URI, CID: payload.CID}, CreatedAt: q.nowISO(), }, q.signer) if err != nil { return out, fmt.Errorf("create the market.event: %w", err) } } return publishedContractEvent{payload: payload, event: envelope}, nil } func (q *Requester) contractEventRecord(event v1alpha1.ContractEvent) (map[string]any, error) { spec := event.Spec switch spec.NSID { case market.ComputeEventsGuestPublicKeyNSID: return market.EncodeRecord(market.ComputeEventsGuestPublicKey{Kind: outboundKind(spec), Tag: spec.Tag, Key: spec.Key, CreatedAt: q.nowISO()}) case market.ComputeEventsGuestSignalWebRTCNSID: return market.EncodeRecord(market.ComputeEventsGuestSignalWebRTC{Kind: spec.Kind, Tag: spec.Tag, Ciphertext: spec.Ciphertext, CreatedAt: q.nowISO()}) case market.ComputeEventsGuestSecretsNSID: return market.EncodeRecord(market.ComputeEventsGuestSecrets{Ciphertext: spec.Ciphertext, CreatedAt: q.nowISO()}) case market.ComputeEventsGuestAgentStartNSID: return market.EncodeRecord(market.ComputeEventsGuestAgentStart{CreatedAt: q.nowISO()}) case market.ComputeEventsGuestSecretsUpdatedNSID: return market.EncodeRecord(market.ComputeEventsGuestSecretsUpdated{CreatedAt: q.nowISO()}) case market.ComputeEventsGuestR2StartNSID: if spec.R2Start == nil { return nil, errors.New("an r2Start event names no spec.r2Start") } return market.EncodeRecord(market.ComputeEventsGuestR2Start{Name: spec.R2Start.Name, Agent: spec.R2Start.Agent, Hold: spec.R2Start.Hold, CreatedAt: q.nowISO()}) case market.ComputeEventsGuestR2ReleaseNSID: return market.EncodeRecord(market.ComputeEventsGuestR2Release{CreatedAt: q.nowISO()}) } return nil, fmt.Errorf("nsid %q is not a guest event", spec.NSID) } func contractEventSubmitDelay(attempts int32) time.Duration { return min(ContractEventSubmitRetry< 0 { if last, err := time.Parse(time.RFC3339, submit.LastAttemptAt); err == nil { if left := last.Add(contractEventSubmitDelay(submit.Attempts)).Sub(now); left > 0 { return left, nil } } } bidderDID := winnerDID(state) bidder, ok := q.winnerBidder(ctx, state, bidderDID) if !ok { bidder, ok = q.bidderForDID(ctx, bidderDID) } var err error code := 0 if !ok || bidder.Endpoint == "" { err = fmt.Errorf("no submitEvent endpoint is known for the bidder %s", bidderDID) } else { submit.Endpoint = bidder.Endpoint submitCtx, cancel := context.WithTimeout(ctx, ContractEventSubmitTimeout) code, err = q.market.SubmitEvent(submitCtx, bidder, SubmitEventRequest{URI: envelope.URI, CID: envelope.CID, Record: envelope.Record}) cancel() if ctx.Err() != nil { return 0, ctx.Err() } } previous := submit.LastError submit.Attempts++ submit.LastAttemptAt = now.Format(time.RFC3339) status.Submit = &submit switch { case err == nil && code >= 200 && code <= 299: submit.SubmittedAt = submit.LastAttemptAt submit.LastError = "" status.Phase = v1alpha1.ContractEventSubmitted status.Message = "submitted to " + bidder.Endpoint conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionSubmitted, metav1.ConditionTrue, "Submitted", status.Message, now) if err := q.writeContractEventStatus(ctx, event, status); err != nil { return 0, err } q.log.Info("requester_contract_event_submitted", "contractEvent", event.Name, "endpoint", bidder.Endpoint, "attempts", submit.Attempts, "event", envelope.URI) return 0, nil case participant.Refused(err): submit.LastError = err.Error() status.Phase = v1alpha1.ContractEventFailed status.Message = "the bidder refused the event: " + submit.LastError conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionSubmitted, metav1.ConditionFalse, "Refused", status.Message, now) q.log.Warn("requester_contract_event_refused_by_bidder", "contractEvent", event.Name, "endpoint", bidder.Endpoint, "status", code, "err", err) return 0, q.writeContractEventStatus(ctx, event, status) case err == nil: submit.LastError = fmt.Sprintf("HTTP %d", code) default: submit.LastError = err.Error() } delay := contractEventSubmitDelay(submit.Attempts) if submit.LastError != previous { q.log.Warn("requester_contract_event_submit_retry", "contractEvent", event.Name, "endpoint", submit.Endpoint, "attempts", submit.Attempts, "retryIn", delay.String(), "err", submit.LastError) } status.Message = "submitting to the bidder: " + submit.LastError conds.SetStatusAt(&status.Conditions, event.Generation, v1alpha1.ContractEventConditionSubmitted, metav1.ConditionFalse, "Retrying", submit.LastError, now) if err := q.writeContractEventStatus(ctx, event, status); err != nil { return 0, err } return delay, nil } func (q *Requester) recordInboundContractEvent(ctx context.Context, contract v1alpha1.ComputeContract, parsed *GuestEvent) error { handle := q.deps.ContractEvents if handle == nil { return nil } name := InboundContractEventName(parsed.Event.URI) spec := v1alpha1.ContractEventSpec{ ContractRef: commonv1.LocalObjectRef{Name: contract.Name}, Direction: v1alpha1.ContractEventInbound, NSID: parsed.Kind, Kind: parsed.GuestKind, Tag: parsed.Tag, Key: parsed.Key, Ciphertext: parsed.Ciphertext, } record := commonv1.StrongRef{URI: parsed.Payload.URI, CID: parsed.Payload.CID} envelope := commonv1.StrongRef{URI: parsed.Event.URI, CID: parsed.Event.CID} existing, known := q.contractEvents.Get(name) if known && existing.Status.Phase == v1alpha1.ContractEventReceived && existing.Status.Record != nil && *existing.Status.Record == record { return nil } q.creatingInbound.Store(name, true) defer q.creatingInbound.Delete(name) if known && !apiequality.Semantic.DeepEqual(existing.Spec, spec) { q.log.Warn("requester_contract_event_replaced", "contractEvent", name, "reason", "its spec does not match the bidder event") if err := handle.DeleteContractEvent(ctx, name); err != nil { return err } return fmt.Errorf("requester: the ContractEvent %s held another spec; it is written again", name) } if !known { err := handle.CreateContractEvent(ctx, v1alpha1.ContractEvent{ ObjectMeta: metav1.ObjectMeta{Name: name, OwnerReferences: []metav1.OwnerReference{contractOwnerReference(contract)}}, Spec: spec, }) if err != nil && !errors.Is(err, contractmarket.ErrExists) { return err } } now := q.now() status := v1alpha1.ContractEventStatus{ Phase: v1alpha1.ContractEventReceived, Message: "received from the bidder " + market.ATURIAuthority(parsed.Event.URI), Record: &record, Event: &envelope, } if known { status.Conditions = existing.Status.DeepCopy().Conditions } stored := v1alpha1.ContractEvent{ObjectMeta: metav1.ObjectMeta{Name: name, Generation: 1, CreationTimestamp: metav1.NewTime(now)}, Spec: spec} if known { stored = existing } conds.SetStatusAt(&status.Conditions, stored.Generation, v1alpha1.ContractEventConditionPublished, metav1.ConditionTrue, "Received", status.Message, now) status.ObservedGeneration = stored.Generation if err := handle.ApplyContractEventStatus(ctx, name, status); err != nil { return err } stored.Status = status q.contractEvents.Put(name, stored) q.log.Info("requester_contract_event_received", "contractEvent", name, "contract", contract.Name, "nsid", parsed.Kind, "tag", parsed.Tag, "record", record.URI) return nil }