tutti/packages/agent/activity-replication/acknowledgement.go

package activityreplication

import (
	"errors"
	"fmt"
)

type AcknowledgementDisposition string

const (
	AcknowledgementApplied   AcknowledgementDisposition = "applied"
	AcknowledgementDuplicate AcknowledgementDisposition = "duplicate"
	AcknowledgementStale     AcknowledgementDisposition = "stale"
)

// MutationAcknowledgement is a sink-side decision used by the shared
// conformance suite. Duplicate and stale are accepted outcomes, not errors.
type MutationAcknowledgement struct {
	MutationID    string                     `json:"mutationId"`
	TransactionID string                     `json:"transactionId"`
	Disposition   AcknowledgementDisposition `json:"disposition"`
	Cursor        uint64                     `json:"cursor"`
}

func AcknowledgeApplied(mutation Mutation, cursor uint64) MutationAcknowledgement {
	return newAcknowledgement(mutation, AcknowledgementApplied, cursor)
}

// AcknowledgeDuplicate preserves the cursor assigned by the original commit.
func AcknowledgeDuplicate(mutation Mutation, originalCursor uint64) MutationAcknowledgement {
	return newAcknowledgement(mutation, AcknowledgementDuplicate, originalCursor)
}

// AcknowledgeStale accepts an obsolete snapshot without writing or assigning a
// cursor, allowing an ordered worker to continue with later mutations.
func AcknowledgeStale(mutation Mutation) MutationAcknowledgement {
	return newAcknowledgement(mutation, AcknowledgementStale, 0)
}

func newAcknowledgement(mutation Mutation, disposition AcknowledgementDisposition, cursor uint64) MutationAcknowledgement {
	return MutationAcknowledgement{
		MutationID:    mutation.MutationID,
		TransactionID: mutation.TransactionID,
		Disposition:   disposition,
		Cursor:        cursor,
	}
}

// SummarizeAcknowledgements produces the wire response for an accepted batch.
func SummarizeAcknowledgements(acknowledgements []MutationAcknowledgement) (ApplyResult, error) {
	result := ApplyResult{}
	for _, acknowledgement := range acknowledgements {
		switch acknowledgement.Disposition {
		case AcknowledgementApplied, AcknowledgementDuplicate, AcknowledgementStale:
		default:
			return ApplyResult{}, fmt.Errorf("unknown activity replication acknowledgement disposition %q", acknowledgement.Disposition)
		}
		result.AcceptedCount++
		if acknowledgement.Cursor > result.Cursor {
			result.Cursor = acknowledgement.Cursor
		}
	}
	return result, nil
}

type RejectionKind string

const (
	RejectionSchema     RejectionKind = "schema"
	RejectionIdentity   RejectionKind = "identity"
	RejectionPermission RejectionKind = "permission"
)

// PermanentRejection identifies a mutation that must be removed or repaired;
// retrying it unchanged cannot succeed.
type PermanentRejection struct {
	Kind          RejectionKind `json:"kind"`
	MutationID    string        `json:"mutationId"`
	TransactionID string        `json:"transactionId"`
	Err           error         `json:"-"`
}

func (e *PermanentRejection) Error() string {
	return fmt.Sprintf("activity replication %s rejection for mutation %q in transaction %q: %v", e.Kind, e.MutationID, e.TransactionID, e.Err)
}

func (e *PermanentRejection) Unwrap() error {
	return e.Err
}

func NewPermanentRejection(kind RejectionKind, mutation Mutation, err error) error {
	if err == nil {
		err = errors.New("activity replication mutation was rejected")
	}
	return &PermanentRejection{
		Kind:          kind,
		MutationID:    mutation.MutationID,
		TransactionID: mutation.TransactionID,
		Err:           err,
	}
}