502 lines
17 KiB
Go
502 lines
17 KiB
Go
package redisactivity
|
|
|
|
import (
|
|
"context"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"strconv"
|
|
"time"
|
|
|
|
"proxy-pool/internal/domain/workerruntime"
|
|
)
|
|
|
|
const runtimeWireVersion = 1
|
|
|
|
const (
|
|
runtimeReplaceSession = "replace_session"
|
|
runtimeCurrentEpoch = "current_epoch"
|
|
runtimeOpenSession = "open_session"
|
|
runtimeValidateSession = "validate_session"
|
|
runtimeRecordSnapshot = "record_snapshot"
|
|
runtimeAcknowledge = "acknowledge_snapshot"
|
|
runtimeReplaceReport = "replace_report"
|
|
runtimeRecordOutcomes = "record_outcomes"
|
|
runtimeRead = "read"
|
|
)
|
|
|
|
type runtimeSessionWire struct {
|
|
Version int `json:"version"`
|
|
WorkerID string `json:"workerId"`
|
|
InstanceID string `json:"instanceId"`
|
|
SessionID string `json:"sessionId"`
|
|
Zone string `json:"zone,omitempty"`
|
|
ProtocolVersion uint32 `json:"protocolVersion,omitempty"`
|
|
Labels map[string]string `json:"labels"`
|
|
AckedSnapshotVersion string `json:"ackedSnapshotVersion"`
|
|
AckedOwnershipEpoch string `json:"ackedOwnershipEpoch"`
|
|
AckedChecksum string `json:"ackedChecksum"`
|
|
RuntimeEnabled bool `json:"runtimeEnabled"`
|
|
}
|
|
|
|
type runtimeSnapshotReferenceWire struct {
|
|
Version int `json:"version"`
|
|
WorkerID string `json:"workerId"`
|
|
SnapshotVersion string `json:"snapshotVersion"`
|
|
OwnershipEpoch string `json:"ownershipEpoch"`
|
|
Checksum string `json:"checksum"`
|
|
}
|
|
|
|
type runtimeSnapshotIssueWire struct {
|
|
Version int `json:"version"`
|
|
SessionID string `json:"sessionId"`
|
|
Reference runtimeSnapshotReferenceWire `json:"reference"`
|
|
}
|
|
|
|
type runtimeSessionValidationWire struct {
|
|
Version int `json:"version"`
|
|
WorkerID string `json:"workerId"`
|
|
SessionID string `json:"sessionId"`
|
|
}
|
|
|
|
type runtimeAcknowledgementWire struct {
|
|
Version int `json:"version"`
|
|
WorkerID string `json:"workerId"`
|
|
SessionID string `json:"sessionId"`
|
|
Reference runtimeSnapshotReferenceWire `json:"reference"`
|
|
Applied bool `json:"applied"`
|
|
ErrorCode string `json:"errorCode"`
|
|
}
|
|
|
|
type runtimeCounterWire struct {
|
|
ProxyID string `json:"proxyId"`
|
|
Active int64 `json:"active"`
|
|
Reserved int64 `json:"reserved"`
|
|
Draining bool `json:"draining"`
|
|
}
|
|
|
|
type runtimeReportWire struct {
|
|
Version int `json:"version"`
|
|
WorkerID string `json:"workerId"`
|
|
SessionID string `json:"sessionId"`
|
|
Sequence string `json:"sequence"`
|
|
SnapshotVersion string `json:"snapshotVersion"`
|
|
OwnershipEpoch string `json:"ownershipEpoch"`
|
|
ObservedAtMS int64 `json:"observedAtMs"`
|
|
Counters []runtimeCounterWire `json:"counters"`
|
|
}
|
|
|
|
type runtimeOutcomeReferenceWire struct {
|
|
Version int `json:"version"`
|
|
WorkerID string `json:"workerId"`
|
|
SessionID string `json:"sessionId"`
|
|
Sequence string `json:"sequence"`
|
|
}
|
|
|
|
type runtimeOwnedProxyWire struct {
|
|
ProxyID string `json:"proxyId"`
|
|
WorkerID string `json:"workerId"`
|
|
OwnershipEpoch string `json:"ownershipEpoch"`
|
|
}
|
|
|
|
type runtimeSnapshotWire struct {
|
|
ProxyID string `json:"proxyId"`
|
|
Active int64 `json:"active"`
|
|
Reserved int64 `json:"reserved"`
|
|
Draining bool `json:"draining"`
|
|
Fresh bool `json:"fresh"`
|
|
}
|
|
|
|
var (
|
|
_ workerruntime.ControlStore = (*Adapter)(nil)
|
|
_ workerruntime.SessionWriter = (*Adapter)(nil)
|
|
_ workerruntime.ReportWriter = (*Adapter)(nil)
|
|
_ workerruntime.RuntimeReader = (*Adapter)(nil)
|
|
_ workerruntime.OutcomeWriter = (*Adapter)(nil)
|
|
)
|
|
|
|
func (a *Adapter) CurrentOwnershipEpoch(ctx context.Context) (uint64, error) {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return 0, err
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeCurrentEpoch, 0, nil, "")
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if reply.Status != scriptOK || reply.Record == "" {
|
|
return 0, invalidScriptReply("unexpected ownership epoch reply")
|
|
}
|
|
epoch, err := strconv.ParseUint(reply.Record, 10, 64)
|
|
if err != nil || epoch == 0 {
|
|
return 0, invalidScriptReply("invalid ownership epoch reply")
|
|
}
|
|
return epoch, nil
|
|
}
|
|
|
|
func (a *Adapter) OpenSession(ctx context.Context, session workerruntime.Session, ttl time.Duration) error {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return err
|
|
}
|
|
normalized, err := workerruntime.NormalizeSession(session)
|
|
if err != nil || ttl <= 0 {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
payload, err := json.Marshal(runtimeSessionWire{
|
|
Version: runtimeWireVersion, WorkerID: normalized.WorkerID, InstanceID: normalized.InstanceID,
|
|
SessionID: normalized.SessionID, Zone: normalized.Zone, ProtocolVersion: normalized.ProtocolVersion,
|
|
Labels: normalized.Labels, AckedSnapshotVersion: "0", AckedOwnershipEpoch: "0",
|
|
AckedChecksum: "", RuntimeEnabled: false,
|
|
})
|
|
if err != nil {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeOpenSession, durationMillis(ttl), payload, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if reply.Status == scriptOK {
|
|
return nil
|
|
}
|
|
if reply.Status == scriptInvalid {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
return invalidScriptReply("unexpected worker open session reply")
|
|
}
|
|
|
|
func (a *Adapter) ValidateSession(ctx context.Context, workerID, sessionID string) error {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return err
|
|
}
|
|
if !workerruntime.ValidIdentifier(workerID) || !workerruntime.ValidIdentifier(sessionID) {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
payload, err := json.Marshal(runtimeSessionValidationWire{
|
|
Version: runtimeWireVersion, WorkerID: workerID, SessionID: sessionID,
|
|
})
|
|
if err != nil {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeValidateSession, 0, payload, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch reply.Status {
|
|
case scriptOK:
|
|
return nil
|
|
case scriptInvalid:
|
|
return workerruntime.ErrInvalidSession
|
|
case scriptUnavailable:
|
|
return workerruntime.ErrStaleSession
|
|
default:
|
|
return invalidScriptReply("unexpected worker session validation reply")
|
|
}
|
|
}
|
|
|
|
func (a *Adapter) RecordIssuedSnapshot(ctx context.Context, sessionID string, reference workerruntime.SnapshotReference, ttl time.Duration) error {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return err
|
|
}
|
|
normalized, err := workerruntime.NormalizeSnapshotReference(reference)
|
|
if err != nil || !workerruntime.ValidIdentifier(sessionID) || ttl <= 0 {
|
|
return workerruntime.ErrInvalidSnapshotReference
|
|
}
|
|
payload, err := json.Marshal(runtimeSnapshotIssueWire{
|
|
Version: runtimeWireVersion, SessionID: sessionID, Reference: referenceWire(normalized),
|
|
})
|
|
if err != nil {
|
|
return workerruntime.ErrInvalidSnapshotReference
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeRecordSnapshot, durationMillis(ttl), payload, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch reply.Status {
|
|
case scriptOK:
|
|
return nil
|
|
case scriptInvalid:
|
|
return workerruntime.ErrInvalidSnapshotReference
|
|
case scriptStale:
|
|
return workerruntime.ErrStaleSnapshotReference
|
|
case scriptConflict:
|
|
return workerruntime.ErrConflictingSnapshotReference
|
|
case scriptSnapshotMismatch:
|
|
return workerruntime.ErrSnapshotMismatch
|
|
case scriptUnavailable:
|
|
return workerruntime.ErrStaleSession
|
|
default:
|
|
return invalidScriptReply("unexpected worker snapshot reference reply")
|
|
}
|
|
}
|
|
|
|
func (a *Adapter) AcknowledgeSnapshot(ctx context.Context, acknowledgement workerruntime.SnapshotAcknowledgement, ttl time.Duration) error {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return err
|
|
}
|
|
normalized, err := workerruntime.NormalizeAcknowledgement(acknowledgement)
|
|
if err != nil || ttl <= 0 {
|
|
return workerruntime.ErrInvalidAcknowledgement
|
|
}
|
|
payload, err := json.Marshal(runtimeAcknowledgementWire{
|
|
Version: runtimeWireVersion, WorkerID: normalized.WorkerID, SessionID: normalized.SessionID,
|
|
Reference: referenceWire(normalized.Reference), Applied: normalized.Applied, ErrorCode: normalized.ErrorCode,
|
|
})
|
|
if err != nil {
|
|
return workerruntime.ErrInvalidAcknowledgement
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeAcknowledge, durationMillis(ttl), payload, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch reply.Status {
|
|
case scriptOK:
|
|
return nil
|
|
case scriptInvalid:
|
|
return workerruntime.ErrInvalidAcknowledgement
|
|
case scriptUnavailable:
|
|
return workerruntime.ErrStaleSession
|
|
case scriptStaleAcknowledgement:
|
|
return workerruntime.ErrStaleAcknowledgement
|
|
case scriptSnapshotMismatch:
|
|
return workerruntime.ErrSnapshotMismatch
|
|
default:
|
|
return invalidScriptReply("unexpected worker acknowledgement reply")
|
|
}
|
|
}
|
|
|
|
func (a *Adapter) ReplaceSession(ctx context.Context, session workerruntime.Session, ttl time.Duration) error {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return err
|
|
}
|
|
if !runtimeClean(session.WorkerID) || !runtimeClean(session.InstanceID) || !runtimeClean(session.SessionID) ||
|
|
session.AckedSnapshotVersion == 0 || session.AckedOwnershipEpoch == 0 || ttl <= 0 {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
payload, err := json.Marshal(runtimeSessionWire{
|
|
Version: runtimeWireVersion, WorkerID: session.WorkerID,
|
|
InstanceID: session.InstanceID, SessionID: session.SessionID,
|
|
AckedSnapshotVersion: strconv.FormatUint(session.AckedSnapshotVersion, 10),
|
|
AckedOwnershipEpoch: strconv.FormatUint(session.AckedOwnershipEpoch, 10),
|
|
RuntimeEnabled: true,
|
|
})
|
|
if err != nil {
|
|
return workerruntime.ErrInvalidSession
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeReplaceSession, durationMillis(ttl), payload, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch reply.Status {
|
|
case scriptOK:
|
|
return nil
|
|
case scriptInvalid:
|
|
return workerruntime.ErrInvalidSession
|
|
case scriptStale:
|
|
return workerruntime.ErrStaleSession
|
|
default:
|
|
return invalidScriptReply("unexpected worker session reply")
|
|
}
|
|
}
|
|
|
|
func (a *Adapter) ReplaceRuntime(ctx context.Context, report workerruntime.Report, ttl time.Duration) error {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return err
|
|
}
|
|
payload, digest, err := a.encodeRuntimeReport(report, ttl)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeReplaceReport, durationMillis(ttl), payload, digest)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch reply.Status {
|
|
case scriptOK:
|
|
return nil
|
|
case scriptInvalid:
|
|
return workerruntime.ErrInvalidReport
|
|
case scriptStale:
|
|
return workerruntime.ErrStaleReport
|
|
case scriptConflict:
|
|
return workerruntime.ErrConflictingReport
|
|
case scriptUnavailable:
|
|
return workerruntime.ErrStaleSession
|
|
case scriptSnapshotMismatch:
|
|
return workerruntime.ErrSnapshotMismatch
|
|
default:
|
|
return invalidScriptReply("unexpected worker runtime reply")
|
|
}
|
|
}
|
|
|
|
// RecordOutcomes persists only the monotonically increasing batch fence. The
|
|
// Gateway event payload is deliberately excluded from Redis to keep the data
|
|
// plane hot path bounded and free of per-request storage writes.
|
|
func (a *Adapter) RecordOutcomes(ctx context.Context, reference workerruntime.OutcomeReference) (uint64, error) {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return 0, err
|
|
}
|
|
normalized, err := workerruntime.NormalizeOutcomeReference(reference)
|
|
if err != nil {
|
|
return 0, workerruntime.ErrInvalidOutcome
|
|
}
|
|
payload, err := json.Marshal(runtimeOutcomeReferenceWire{
|
|
Version: runtimeWireVersion, WorkerID: normalized.WorkerID, SessionID: normalized.SessionID,
|
|
Sequence: strconv.FormatUint(normalized.Sequence, 10),
|
|
})
|
|
if err != nil {
|
|
return 0, workerruntime.ErrInvalidOutcome
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeRecordOutcomes, 0, payload, hex.EncodeToString(normalized.Digest[:]))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
switch reply.Status {
|
|
case scriptOK:
|
|
accepted, parseErr := strconv.ParseUint(reply.Record, 10, 64)
|
|
if parseErr != nil || accepted == 0 {
|
|
return 0, invalidScriptReply("invalid worker outcome reply")
|
|
}
|
|
return accepted, nil
|
|
case scriptInvalid:
|
|
return 0, workerruntime.ErrInvalidOutcome
|
|
case scriptStale:
|
|
return 0, workerruntime.ErrStaleOutcome
|
|
case scriptConflict:
|
|
return 0, workerruntime.ErrConflictingOutcome
|
|
case scriptUnavailable:
|
|
return 0, workerruntime.ErrStaleSession
|
|
default:
|
|
return 0, invalidScriptReply("unexpected worker outcome reply")
|
|
}
|
|
}
|
|
|
|
func (a *Adapter) ReadRuntime(ctx context.Context, proxies []workerruntime.OwnedProxy) ([]workerruntime.Snapshot, error) {
|
|
if err := validateRuntimeCall(ctx, a); err != nil {
|
|
return nil, err
|
|
}
|
|
if len(proxies) > a.options.MaxRuntimeCounters {
|
|
return nil, workerruntime.ErrInvalidQuery
|
|
}
|
|
wires := make([]runtimeOwnedProxyWire, len(proxies))
|
|
seen := make(map[string]struct{}, len(proxies))
|
|
for index, proxy := range proxies {
|
|
if !workerruntime.ValidIdentifier(proxy.ProxyID) || !workerruntime.ValidIdentifier(proxy.WorkerID) || proxy.OwnershipEpoch == 0 {
|
|
return nil, workerruntime.ErrInvalidQuery
|
|
}
|
|
key := proxy.WorkerID + "\x00" + proxy.ProxyID
|
|
if _, exists := seen[key]; exists {
|
|
return nil, workerruntime.ErrInvalidQuery
|
|
}
|
|
seen[key] = struct{}{}
|
|
wires[index] = runtimeOwnedProxyWire{
|
|
ProxyID: proxy.ProxyID, WorkerID: proxy.WorkerID,
|
|
OwnershipEpoch: strconv.FormatUint(proxy.OwnershipEpoch, 10),
|
|
}
|
|
}
|
|
payload, err := json.Marshal(wires)
|
|
if err != nil {
|
|
return nil, workerruntime.ErrInvalidQuery
|
|
}
|
|
reply, err := a.runRuntime(ctx, runtimeRead, 0, payload, "")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if reply.Status == scriptInvalid {
|
|
return nil, workerruntime.ErrInvalidQuery
|
|
}
|
|
if reply.Status != scriptOK || len(reply.Snapshots) != len(proxies) {
|
|
return nil, invalidScriptReply("unexpected worker runtime read reply")
|
|
}
|
|
result := make([]workerruntime.Snapshot, len(reply.Snapshots))
|
|
for index, snapshot := range reply.Snapshots {
|
|
if snapshot.ProxyID != proxies[index].ProxyID || snapshot.Active < 0 || snapshot.Reserved < 0 {
|
|
return nil, invalidScriptReply("invalid worker runtime snapshot")
|
|
}
|
|
result[index] = workerruntime.Snapshot{
|
|
ProxyID: snapshot.ProxyID, Active: snapshot.Active, Reserved: snapshot.Reserved,
|
|
Draining: snapshot.Draining, Fresh: snapshot.Fresh,
|
|
}
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func (a *Adapter) encodeRuntimeReport(report workerruntime.Report, ttl time.Duration) ([]byte, string, error) {
|
|
normalized, digest, err := workerruntime.NormalizeReport(report)
|
|
if err != nil || ttl <= 0 || len(normalized.Counters) > a.options.MaxRuntimeCounters {
|
|
return nil, "", workerruntime.ErrInvalidReport
|
|
}
|
|
wires := make([]runtimeCounterWire, len(normalized.Counters))
|
|
for index, counter := range normalized.Counters {
|
|
wires[index] = runtimeCounterWire{
|
|
ProxyID: counter.ProxyID, Active: counter.Active,
|
|
Reserved: counter.Reserved, Draining: counter.Draining,
|
|
}
|
|
}
|
|
payload, err := json.Marshal(runtimeReportWire{
|
|
Version: runtimeWireVersion, WorkerID: normalized.WorkerID, SessionID: normalized.SessionID,
|
|
Sequence: strconv.FormatUint(normalized.Sequence, 10),
|
|
SnapshotVersion: strconv.FormatUint(normalized.SnapshotVersion, 10),
|
|
OwnershipEpoch: strconv.FormatUint(normalized.OwnershipEpoch, 10),
|
|
ObservedAtMS: normalized.ObservedAt.UnixMilli(), Counters: wires,
|
|
})
|
|
if err != nil {
|
|
return nil, "", workerruntime.ErrInvalidReport
|
|
}
|
|
return payload, hex.EncodeToString(digest[:]), nil
|
|
}
|
|
|
|
func (a *Adapter) runRuntime(
|
|
ctx context.Context,
|
|
operation string,
|
|
ttlMS int64,
|
|
payload []byte,
|
|
digest string,
|
|
) (runtimeScriptReply, error) {
|
|
result, err := runScript(ctx, a.client, runtimeScript, []string{
|
|
a.keys.workerSessions, a.keys.workerSessionExpiry,
|
|
a.keys.workerSnapshots, a.keys.workerSnapshotExpiry,
|
|
a.keys.workerRuntime, a.keys.workerRuntimeExpiry, a.keys.owners, a.keys.epoch,
|
|
a.keys.workerOutcomes, a.keys.records, a.keys.available, a.keys.ownerExpiry,
|
|
a.keys.drainTickets, a.keys.workerDraining(runtimeWorkerID(payload)),
|
|
}, operation, ttlMS, a.options.CleanupLimit, string(payload), digest)
|
|
if err != nil {
|
|
return runtimeScriptReply{}, err
|
|
}
|
|
var reply runtimeScriptReply
|
|
if err := decodeScriptResult(result, &reply); err != nil {
|
|
return runtimeScriptReply{}, err
|
|
}
|
|
return reply, nil
|
|
}
|
|
|
|
func runtimeWorkerID(payload []byte) string {
|
|
var value struct {
|
|
WorkerID string `json:"workerId"`
|
|
}
|
|
if json.Unmarshal(payload, &value) != nil {
|
|
return ""
|
|
}
|
|
return value.WorkerID
|
|
}
|
|
|
|
func validateRuntimeCall(ctx context.Context, adapter *Adapter) error {
|
|
if ctx == nil || adapter == nil {
|
|
return workerruntime.ErrInvalidStore
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func referenceWire(reference workerruntime.SnapshotReference) runtimeSnapshotReferenceWire {
|
|
return runtimeSnapshotReferenceWire{
|
|
Version: runtimeWireVersion, WorkerID: reference.WorkerID,
|
|
SnapshotVersion: strconv.FormatUint(reference.Version, 10),
|
|
OwnershipEpoch: strconv.FormatUint(reference.OwnershipEpoch, 10),
|
|
Checksum: hex.EncodeToString(reference.Checksum[:]),
|
|
}
|
|
}
|
|
|
|
func runtimeClean(value string) bool {
|
|
return workerruntime.ValidIdentifier(value)
|
|
}
|