312 lines
12 KiB
Go
312 lines
12 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"crypto/subtle"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"slices"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
|
platform "git.yuntuo.hk/k2safe/orbit/platform/model"
|
|
)
|
|
|
|
// materializeResults 持有订阅锁，流和日志只做已提交读，不反向锁生产者。
|
|
// 先保存独立交付记录再推进扫描序号；未确认窗口满时停止扫描，保留来源日志。
|
|
func materializeResults(ctx context.Context, tx *sql.Tx, ns core.ID, sub *events.Subscription) error {
|
|
if sub.PendingCount >= sub.MaxPending {
|
|
return nil
|
|
}
|
|
var head, retained uint64
|
|
if err := tx.QueryRowContext(ctx, `SELECT last_sequence,retained_from_sequence FROM orbit_event_streams WHERE namespace_id=?`, ns).Scan(&head, &retained); err != nil {
|
|
return err
|
|
}
|
|
if sub.CursorSequence < retained-1 {
|
|
return events.HistoryExpired
|
|
}
|
|
limit := min(uint32(100), sub.MaxPending-sub.PendingCount)
|
|
rows, err := tx.QueryContext(ctx, `SELECT r.id,r.sequence,e.type FROM orbit_result_events r LEFT JOIN orbit_events e ON e.namespace_id=r.namespace_id AND e.id=r.id WHERE r.namespace_id=? AND r.queue_id=? AND r.sequence>? AND r.sequence<=? ORDER BY r.sequence LIMIT ?`, ns, sub.QueueID, sub.CursorSequence, head, limit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
type candidate struct {
|
|
id core.ID
|
|
sequence uint64
|
|
eventType string
|
|
}
|
|
batch := make([]candidate, 0, limit)
|
|
for rows.Next() {
|
|
var c candidate
|
|
if err = rows.Scan(&c.id, &c.sequence, &c.eventType); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
batch = append(batch, c)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
cursor := sub.CursorSequence
|
|
pending := sub.PendingCount
|
|
for _, c := range batch {
|
|
cursor = c.sequence
|
|
if !slices.Contains(sub.Types, c.eventType) {
|
|
continue
|
|
}
|
|
id, err := core.NewID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `INSERT INTO orbit_deliveries(namespace_id,id,subscription_id,event_id,available_at,attempt_budget) VALUES (?,?,?,?,UTC_TIMESTAMP(6),?)`, ns, id, sub.ID, c.id, sub.MaxAttempts); err != nil {
|
|
return err
|
|
}
|
|
pending++
|
|
}
|
|
if len(batch) < int(limit) {
|
|
cursor = head
|
|
}
|
|
if cursor == sub.CursorSequence && pending == sub.PendingCount {
|
|
return nil
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET cursor_sequence=?,pending_count=?,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, cursor, pending, ns, sub.ID); err != nil {
|
|
return err
|
|
}
|
|
sub.CursorSequence, sub.PendingCount = cursor, pending
|
|
return nil
|
|
}
|
|
|
|
// recoverDeliveries 在新领取前有界回收过期租约。耗尽预算仍占积压，禁止静默丢弃。
|
|
// 最后一次证明保留；迟到确认只在还没有新交付代次时有效。
|
|
func recoverDeliveries(ctx context.Context, tx *sql.Tx, ns, sub core.ID, now time.Time) error {
|
|
_, err := tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state=IF(attempts>=attempt_budget,'EXPIRED','PENDING'),available_at=TIMESTAMPADD(SECOND,LEAST(30,POW(2,LEAST(attempts-1,5))),?),lease_owner=NULL,lease_until=NULL,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND state='LEASED' AND lease_until<=?`, now, now, ns, sub, now)
|
|
return err
|
|
}
|
|
|
|
type storedDelivery struct {
|
|
view events.Delivery
|
|
digest []byte
|
|
}
|
|
|
|
func readDelivery(ctx context.Context, tx *sql.Tx, ns, sub, id core.ID, lock bool) (storedDelivery, error) {
|
|
var out storedDelivery
|
|
q := `SELECT id,subscription_id,event_id,revision,state,available_at,lease_epoch,lease_until,attempts,attempt_budget,acknowledged_at,token_digest,created_at FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND id=?`
|
|
if lock {
|
|
q += ` FOR UPDATE`
|
|
}
|
|
v := &out.view
|
|
err := tx.QueryRowContext(ctx, q, ns, sub, id).Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &out.digest, &v.CreatedAt)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return out, fault.NotFound
|
|
}
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
err = tx.QueryRowContext(ctx, `SELECT sequence FROM orbit_result_events WHERE namespace_id=? AND id=?`, ns, v.EventID).Scan(&v.Sequence)
|
|
return out, err
|
|
}
|
|
func readResultEvent(ctx context.Context, tx *sql.Tx, ns, id core.ID) (events.Event, error) {
|
|
var out events.Event
|
|
var details, snapshot []byte
|
|
err := tx.QueryRowContext(ctx, `SELECT r.id,r.sequence,e.type,e.aggregate_kind,e.aggregate_id,e.aggregate_revision,e.occurred_at,e.payload,r.snapshot FROM orbit_result_events r LEFT JOIN orbit_events e ON e.namespace_id=r.namespace_id AND e.id=r.id WHERE r.namespace_id=? AND r.id=?`, ns, id).Scan(&out.ID, &out.Sequence, &out.Type, &out.AggregateKind, &out.AggregateID, &out.AggregateRevision, &out.OccurredAt, &details, &snapshot)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
out.Details = details
|
|
err = json.Unmarshal(snapshot, &out.Snapshot)
|
|
return out, err
|
|
}
|
|
func (s *EventStore) GetDelivery(ctx context.Context, t auth.Token, sub, id core.ID) (events.Delivery, error) {
|
|
var out events.Delivery
|
|
err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, err := readSubscription(ctx, tx, t, p, sub, false); err != nil {
|
|
return err
|
|
}
|
|
d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
|
|
out = d.view
|
|
return err
|
|
})
|
|
return out, err
|
|
}
|
|
func (s *EventStore) ClaimDelivery(ctx context.Context, t auth.Token, id core.ID, in events.ClaimInput) (*events.Grant, error) {
|
|
var out *events.Grant
|
|
err := s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
|
|
sub, err := readSubscription(ctx, tx, t, p, id, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !sub.Enabled {
|
|
return nil
|
|
}
|
|
now, err := dbNow(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = recoverDeliveries(ctx, tx, t.NamespaceID, id, now); err != nil {
|
|
return err
|
|
}
|
|
if err = materializeResults(ctx, tx, t.NamespaceID, &sub); err != nil {
|
|
return err
|
|
}
|
|
// 物化记录的 available_at 来自写入时的数据库时钟；不能继续用物化前的时间领取。
|
|
now, err = dbNow(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var active uint32
|
|
if err = tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='LEASED' AND lease_owner=?`, t.NamespaceID, id, in.ConnectionID).Scan(&active); err != nil {
|
|
return err
|
|
}
|
|
if active >= in.Capacity {
|
|
return nil
|
|
}
|
|
var deliveryID core.ID
|
|
err = tx.QueryRowContext(ctx, `SELECT id FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state='PENDING' AND available_at<=? ORDER BY available_at,id LIMIT 1 FOR UPDATE`, t.NamespaceID, id, now).Scan(&deliveryID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
stored, err := readDelivery(ctx, tx, t.NamespaceID, id, deliveryID, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
v := stored.view
|
|
if v.Attempts >= v.AttemptBudget || v.LeaseEpoch == math.MaxUint64 || v.Revision == math.MaxUint64 {
|
|
return fault.Conflict
|
|
}
|
|
secret, digest, err := events.NewSecret()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
until := now.Add(events.LeaseDuration)
|
|
if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='LEASED',lease_owner=?,lease_epoch=lease_epoch+1,lease_until=?,token_digest=?,attempts=attempts+1,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND id=?`, in.ConnectionID, until, digest, now, t.NamespaceID, id, deliveryID); err != nil {
|
|
return err
|
|
}
|
|
v.State = platform.DeliveryLeased
|
|
v.LeaseEpoch++
|
|
v.LeaseUntil = &until
|
|
v.Attempts++
|
|
v.Revision++
|
|
event, err := readResultEvent(ctx, tx, t.NamespaceID, v.EventID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
operation, err := core.NewID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAudit(ctx, tx, t, operation, v.ID, "delivery", "delivery.lease", "grant durable result delivery", nil, map[string]any{"event_id": v.EventID, "epoch": fmt.Sprint(v.LeaseEpoch), "attempts": v.Attempts, "lease_until": until, "connection_id": in.ConnectionID}); err != nil {
|
|
return err
|
|
}
|
|
out = &events.Grant{Delivery: v, Event: event, LeaseSecret: secret}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return out, nil
|
|
}
|
|
func (s *EventStore) AckDelivery(ctx context.Context, t auth.Token, id core.ID, in events.AckInput) (events.AckResult, error) {
|
|
var out events.AckResult
|
|
digest, err := events.SecretDigest(in.LeaseSecret)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
err = s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
|
|
sub, err := readSubscription(ctx, tx, t, p, id, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
d, err := readDelivery(ctx, tx, t.NamespaceID, id, in.DeliveryID, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
v := d.view
|
|
if v.LeaseEpoch != in.Epoch || subtle.ConstantTimeCompare(d.digest, digest[:]) != 1 {
|
|
return events.LeaseLost
|
|
}
|
|
out = events.AckResult{DeliveryID: v.ID, EventID: v.EventID, Epoch: in.Epoch}
|
|
if v.State == platform.DeliveryAcked {
|
|
if v.AcknowledgedAt == nil {
|
|
return errors.New("acked delivery lacks confirmation timestamp")
|
|
}
|
|
out.AcknowledgedAt = *v.AcknowledgedAt
|
|
out.Deduplicated = true
|
|
return nil
|
|
}
|
|
if sub.PendingCount == 0 || v.Revision == math.MaxUint64 {
|
|
return errors.New("delivery confirmation invariant failed")
|
|
}
|
|
now, err := dbNow(ctx, tx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='ACKED',lease_owner=NULL,lease_until=NULL,acknowledged_at=?,revision=revision+1,updated_at=? WHERE namespace_id=? AND subscription_id=? AND id=?`, now, now, t.NamespaceID, id, v.ID); err != nil {
|
|
return err
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET pending_count=pending_count-1,updated_at=? WHERE namespace_id=? AND id=?`, now, t.NamespaceID, id); err != nil {
|
|
return err
|
|
}
|
|
operation, err := core.NewID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAudit(ctx, tx, t, operation, v.ID, "delivery", "delivery.ack", "consumer confirmed durable processing", map[string]any{"state": v.State}, map[string]any{"state": platform.DeliveryAcked, "event_id": v.EventID, "epoch": fmt.Sprint(in.Epoch)}); err != nil {
|
|
return err
|
|
}
|
|
out.AcknowledgedAt = now
|
|
return nil
|
|
})
|
|
return out, err
|
|
}
|
|
func (s *EventStore) RetryDelivery(ctx context.Context, t auth.Token, sub, id core.ID, key string, in events.RetryInput) (events.RetryResult, error) {
|
|
var out events.RetryResult
|
|
err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error {
|
|
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "delivery.retry", eventScope(t.CredentialID, sub, id), key, "delivery", id, in)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = readSubscription(ctx, tx, t, p, sub, true); err != nil {
|
|
return err
|
|
}
|
|
if cached != nil {
|
|
if err = json.Unmarshal(cached, &out); err != nil {
|
|
return err
|
|
}
|
|
out.Deduplicated = true
|
|
return nil
|
|
}
|
|
d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
v := d.view
|
|
if v.State != platform.DeliveryExpired || v.Revision != in.ExpectedRevision || v.Revision == math.MaxUint64 || uint64(v.AttemptBudget)+uint64(in.AdditionalAttempts) > 1000000 {
|
|
return fault.Conflict
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE orbit_deliveries SET state='PENDING',attempt_budget=attempt_budget+?,available_at=UTC_TIMESTAMP(6),revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND subscription_id=? AND id=?`, in.AdditionalAttempts, t.NamespaceID, sub, id); err != nil {
|
|
return err
|
|
}
|
|
d, err = readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
out.Delivery = d.view
|
|
if err = appendAudit(ctx, tx, t, identity, id, "delivery", "delivery.retry", in.Reason, v, out.Delivery); err != nil {
|
|
return err
|
|
}
|
|
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
|
})
|
|
return out, err
|
|
}
|