Skip to content

cmek split 6/8: eventstore encrypted persistence #4573

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ti-chi-bot merged 23 commits into from
May 12, 2026
Merged

cmek split 6/8: eventstore encrypted persistence #4573

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
c7b64fe
feat(eventstore): support encrypted persisted events
tenfyzhong Mar 23, 2026
9aafb69
feat(eventstore): remove encryption error logging
tenfyzhong Apr 29, 2026
f55924a
eventstore: reorder write branches for fast path
tenfyzhong May 6, 2026
c1020d0
eventstore: refactor write value preparation
tenfyzhong May 6, 2026
dcb834c
Revert "eventstore: refactor write value preparation"
tenfyzhong May 6, 2026
c5d957a
eventstore,encryption: frame plaintext encryption-layer values
tenfyzhong May 6, 2026
2d9068e
eventstore,encryption: preserve legacy framed value reads
tenfyzhong May 6, 2026
70b86b4
eventstore: restore encryption-layer key metadata
tenfyzhong May 7, 2026
b11a330
Merge remote-tracking branch 'origin/master' into split/pr3955-06-eve…
tenfyzhong May 7, 2026
c973a71
eventstore: deduplicate writeEvents encoding
tenfyzhong May 7, 2026
a2d8300
eventstore: widen encryption-layer key mask
tenfyzhong May 7, 2026
33733ad
eventstore: simplify key attribute encoding
tenfyzhong May 7, 2026
b776bef
eventstore: clarify key field offsets
tenfyzhong May 7, 2026
3da9b73
fix(eventstore): Correct diagram alignment in format.go
tenfyzhong May 7, 2026
67be557
eventstore: clarify key field lengths
tenfyzhong May 7, 2026
38af250
eventstore: move key layout comment
tenfyzhong May 7, 2026
3f3e359
encryption: tighten encryption layer decoding
tenfyzhong May 7, 2026
3fa8a5b
eventstore: clarify key boundary lengths
tenfyzhong May 7, 2026
ab781e5
logservice: remove unused eventstore raw value return
tenfyzhong May 8, 2026
a43786a
pkg/encryption: remove decode panic tests
tenfyzhong May 9, 2026
14d624b
Merge branch 'master' into split/pr3955-06-eventstore
tenfyzhong May 9, 2026
9ba7503
Merge branch 'master' into split/pr3955-06-eventstore
tenfyzhong May 11, 2026
78b971f
tests: move debezium_basic to heavy integration tests
tenfyzhong May 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
191 changes: 138 additions & 53 deletions logservice/eventstore/event_store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import (
"github.com/pingcap/ticdc/pkg/common"
appcontext "github.com/pingcap/ticdc/pkg/common/context"
"github.com/pingcap/ticdc/pkg/config"
"github.com/pingcap/ticdc/pkg/encryption"
"github.com/pingcap/ticdc/pkg/messaging"
"github.com/pingcap/ticdc/pkg/metrics"
"github.com/pingcap/ticdc/pkg/node"
Expand Down Expand Up @@ -118,6 +119,8 @@ type dispatcherStat struct {
resolvedTs atomic.Uint64
// the max ts of events which is not needed by this dispatcher
checkpointTs uint64
// keyspaceID for encryption (0 means default/classic)
keyspaceID uint32
// the difference between `subStat`, `pendingSubStat` and `removingSubStat`:
// 1) if there is no existing subscriptions which can be reused,
// or there is a existing subscription with exact span match,
Expand Down Expand Up @@ -184,9 +187,10 @@ type subscriptionStat struct {
type subscriptionStats map[logpuller.SubscriptionID]*subscriptionStat

type eventWithCallback struct {
subID logpuller.SubscriptionID
tableID int64
kvs []common.RawKVEntry
subID logpuller.SubscriptionID
tableID int64
keyspaceID uint32
kvs []common.RawKVEntry
// kv with commitTs <= currentResolvedTs will be filtered out
currentResolvedTs uint64
enqueueTimeNano int64
Expand Down Expand Up @@ -242,6 +246,8 @@ type eventStore struct {
compressionThreshold int
// enableZstdCompression controls whether to enable zstd compression for large values.
enableZstdCompression bool
// encryptionManager for encrypting/decrypting data (optional).
encryptionManager encryption.EncryptionManager
}

const (
Expand All @@ -262,6 +268,8 @@ func New(
log.Panic("fail to remove path", zap.String("path", dbPath), zap.Error(err))
}

// Try to get encryption manager from appcontext (optional)
encMgr, _ := appcontext.TryGetService[encryption.EncryptionManager](appcontext.EncryptionManager)
dbs, pebbleCache, tableCache := createPebbleDBs(dbPath, dbCount)
store := &eventStore{
pdClock: appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock),
Expand All @@ -287,6 +295,7 @@ func New(
},
compressionThreshold: config.GetGlobalServerConfig().Debug.EventStore.CompressionThreshold,
enableZstdCompression: config.GetGlobalServerConfig().Debug.EventStore.EnableZstdCompression,
encryptionManager: encMgr,
}
store.gcManager = newGCManager(store.dbs, deleteDataRange, compactDataRange)

Expand Down Expand Up @@ -501,6 +510,7 @@ func (e *eventStore) RegisterDispatcher(
dispatcherID: dispatcherID,
tableSpan: dispatcherSpan,
checkpointTs: startTs,
keyspaceID: dispatcherSpan.KeyspaceID,
}
stat.resolvedTs.Store(startTs)

Expand Down Expand Up @@ -627,6 +637,7 @@ func (e *eventStore) RegisterDispatcher(
subStat.eventCh.Push(eventWithCallback{
subID: subStat.subID,
tableID: subStat.tableSpan.TableID,
keyspaceID: subStat.tableSpan.KeyspaceID,

@coderabbitai coderabbitai Bot May 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Confirm whether reuse logic anywhere filters by KeyspaceID and how tableStats are keyed.
rg -nP -C3 'tableStats\[' --type=go
rg -nP -C3 'KeyspaceID' logservice/eventstore/
ast-grep --pattern $'if bytes.Compare(subStat.tableSpan.StartKey, $_) <= 0 && $$$ { $$$ }'

Repository: pingcap/ticdc

Length of output: 20144

🏁 Script executed:

#!/bin/bash
# Look at read path and iterator implementation around line 915-928
sed -n '900,950p' logservice/eventstore/event_store.go
# Search for iterator and decryption usage
rg -nP 'keyspaceID' logservice/eventstore/event_store.go -A 2 -B 2 | grep -A 5 -B 5 'iterator\|Decrypt\|Next'
# Look at the iterator's Next() implementation
ast-grep --pattern 'func ($$_) Next() {
  $$$
}'

Repository: pingcap/ticdc

Length of output: 1966

Keyspace ID mismatch in subscription reuse allows decryption to fail and panic.

The reuse logic in RegisterDispatcher (lines 509–566) matches existing subscriptions only by TableID and span containment, ignoring KeyspaceID. This allows a dispatcher to attach to a subStat with a different KeyspaceID.

When this happens:

  • Write path (line 624): encrypts with subStat.tableSpan.KeyspaceID (subscription's keyspace)
  • Read path (line 927, 1525): decrypts with stat.keyspaceID (dispatcher's keyspace)

If the keyspaces differ, DecryptData fails and the iterator panics at line 1527. The test only covers the matching case, leaving this silently exploitable in production with EnableDataSharing enabled.

Fix either by:

  1. Refusing to reuse a subStat with different KeyspaceID, or
  2. Decrypting with the subscription's keyspaceID (subStat.tableSpan.KeyspaceID) to match the write side.
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@logservice/eventstore/event_store.go` at line 624, The subscription-reuse
logic in RegisterDispatcher must ensure keyspace equality before attaching a new
dispatcher to an existing subStat: update the matching condition that currently
checks TableID and span containment to also require subStat.tableSpan.KeyspaceID
== tableSpan.KeyspaceID (the incoming span/keyspace); if they differ, treat it
as non-matching and create a new subStat instead of reusing. This prevents
writes encrypted with subStat.tableSpan.KeyspaceID from being decrypted later
with a different stat.keyspaceID (used by DecryptData) and avoids the iterator
panic.

kvs: kvs,
currentResolvedTs: subStat.resolvedTs.Load(),
enqueueTimeNano: now.UnixNano(),
Expand Down Expand Up @@ -951,16 +962,18 @@ func (e *eventStore) GetIterator(dispatcherID common.DispatcherID, dataRange com
}

return &eventStoreIter{
tableSpan: stat.tableSpan,
needCheckSpan: needCheckSpan,
innerIter: iter,
prevStartTs: 0,
prevCommitTs: 0,
startTs: dataRange.CommitTsStart,
endTs: dataRange.CommitTsEnd,
rowCount: 0,
decoder: decoder,
decoderPool: e.decoderPool,
tableSpan: stat.tableSpan,
needCheckSpan: needCheckSpan,
innerIter: iter,
prevStartTs: 0,
prevCommitTs: 0,
startTs: dataRange.CommitTsStart,
endTs: dataRange.CommitTsEnd,
rowCount: 0,
decoder: decoder,
decoderPool: e.decoderPool,
encryptionManager: e.encryptionManager,
keyspaceID: stat.keyspaceID,
}, nil
}

Expand Down Expand Up @@ -1364,37 +1377,62 @@ func (e *eventStore) writeEvents(
continue
}

compressionType := CompressionNone
valueBytesBefore := kv.GetSize()
valueBytesAfter := valueBytesBefore
keyLen := encodedKeyLen(kv)
compressionType := CompressionNone
needCompress := e.enableZstdCompression && valueBytesBefore > int64(e.compressionThreshold)
if e.encryptionManager == nil && !needCompress {
// SetDeferred reserves the final Pebble batch space up front, so
// the uncompressed raw KV can be encoded directly into op.Value
// without allocating a separate value slice and copying it later.
op := batch.SetDeferred(keyLen, int(valueBytesBefore))
if err := encodeDeferredEventKey(op, keyLen, uint64(event.subID), event.tableID, kv, CompressionNone, false); err != nil {
return err
}
op.Value = kv.EncodeTo(op.Value[:0])
if len(op.Value) != int(valueBytesBefore) {
return fmt.Errorf("encoded raw kv entry size mismatch, expected %d, got %d",
valueBytesBefore, len(op.Value))
}
if err := op.Finish(); err != nil {
return err
}
} else if e.encryptionManager != nil {
var value []byte
value, compressionType, rawBuf, dstBuf = encodeAndMaybeCompressValue(kv, encoder, rawBuf, dstBuf, needCompress)
valueBytesAfter = int64(len(value))

if e.enableZstdCompression && valueBytesBefore > int64(e.compressionThreshold) {
if cap(rawBuf) < int(valueBytesBefore) {
rawBuf = make([]byte, 0, int(valueBytesBefore))
} else {
rawBuf = rawBuf[:0]
// Encrypt if encryption is enabled (after compression)
encryptedValue, err := e.encryptionManager.EncryptData(context.Background(), event.keyspaceID, value)
if err != nil {
Comment thread
coderabbitai[bot] marked this conversation as resolved.
return err
}
rawValue := kv.EncodeTo(rawBuf)
maxEncodedSize := encoder.MaxEncodedSize(len(rawValue))
if cap(dstBuf) < maxEncodedSize {
dstBuf = make([]byte, 0, maxEncodedSize)
} else {
dstBuf = dstBuf[:0]

op := batch.SetDeferred(keyLen, len(encryptedValue))
if err := encodeDeferredEventKey(op, keyLen, uint64(event.subID), event.tableID, kv, compressionType, true); err != nil {
return err
}
value := encoder.EncodeAll(rawValue, dstBuf)
copiedValueLen := copy(op.Value, encryptedValue)
op.Value = op.Value[:copiedValueLen]
if copiedValueLen != len(encryptedValue) {
return fmt.Errorf("encrypted raw kv entry size mismatch, expected %d, got %d",
len(encryptedValue), copiedValueLen)
}
if err := op.Finish(); err != nil {
return err
}
} else {
var value []byte
value, compressionType, rawBuf, dstBuf = encodeAndMaybeCompressValue(kv, encoder, rawBuf, dstBuf, true)
valueBytesAfter = int64(len(value))
compressionType = CompressionZSTD
metrics.EventStoreCompressedRowsCount.Inc()
// SetDeferred is a write path optimization. Now that the compressed
// value length is known, reserve the exact key/value space in the
// Pebble batch, encode the key directly into op.Key, and copy the
// compressed value into op.Value without building a temporary key.
op := batch.SetDeferred(keyLen, len(value))
op.Key = EncodeKeyTo(op.Key[:0], uint64(event.subID), event.tableID, kv, compressionType)
if len(op.Key) != keyLen {
return fmt.Errorf("encoded event store key size mismatch, expected %d, got %d",
keyLen, len(op.Key))
if err := encodeDeferredEventKey(op, keyLen, uint64(event.subID), event.tableID, kv, compressionType, false); err != nil {
return err
}
copiedValueLen := copy(op.Value, value)
op.Value = op.Value[:copiedValueLen]
Expand All @@ -1405,28 +1443,7 @@ func (e *eventStore) writeEvents(
if err := op.Finish(); err != nil {
return err
}
rawBuf = rawValue[:0]
dstBuf = value[:0]
} else {
// SetDeferred reserves the final Pebble batch space up front, so
// the uncompressed raw KV can be encoded directly into op.Value
// without allocating a separate value slice and copying it later.
op := batch.SetDeferred(keyLen, int(valueBytesBefore))
op.Key = EncodeKeyTo(op.Key[:0], uint64(event.subID), event.tableID, kv, compressionType)
if len(op.Key) != keyLen {
return fmt.Errorf("encoded event store key size mismatch, expected %d, got %d",
keyLen, len(op.Key))
}
op.Value = kv.EncodeTo(op.Value[:0])
if len(op.Value) != int(valueBytesBefore) {
return fmt.Errorf("encoded raw kv entry size mismatch, expected %d, got %d",
valueBytesBefore, len(op.Value))
}
if err := op.Finish(); err != nil {
return err
}
}

totalValueBytesBefore += valueBytesBefore
totalValueBytesAfter += valueBytesAfter
}
Expand All @@ -1451,6 +1468,60 @@ func (e *eventStore) writeEvents(
return err
}

func encodeAndMaybeCompressValue(
kv *common.RawKVEntry,
encoder *zstd.Encoder,
rawBuf []byte,
dstBuf []byte,
needCompress bool,
) (value []byte, compressionType CompressionType, nextRawBuf []byte, nextDstBuf []byte) {
rawValue := ensureValueBuffer(rawBuf, int(kv.GetSize()))
rawValue = kv.EncodeTo(rawValue)
value = rawValue
compressionType = CompressionNone
nextRawBuf = rawValue[:0]
nextDstBuf = dstBuf
if !needCompress {
return value, compressionType, nextRawBuf, nextDstBuf
}

maxEncodedSize := encoder.MaxEncodedSize(len(rawValue))
value = ensureValueBuffer(dstBuf, maxEncodedSize)
value = encoder.EncodeAll(rawValue, value)
compressionType = CompressionZSTD
nextDstBuf = value[:0]
metrics.EventStoreCompressedRowsCount.Inc()
return value, compressionType, nextRawBuf, nextDstBuf
}

func ensureValueBuffer(buf []byte, minCap int) []byte {
if cap(buf) < minCap {
return make([]byte, 0, minCap)
}
return buf[:0]
}

func encodeDeferredEventKey(
op *pebble.DeferredBatchOp,
keyLen int,
subID uint64,
tableID int64,
kv *common.RawKVEntry,
compressionType CompressionType,
usesEncryptionLayer bool,
) error {
if usesEncryptionLayer {
op.Key = encodeKeyToWithEncryptionLayer(op.Key[:0], subID, tableID, kv, compressionType)
} else {
op.Key = EncodeKeyTo(op.Key[:0], subID, tableID, kv, compressionType)
}
if len(op.Key) != keyLen {
return fmt.Errorf("encoded event store key size mismatch, expected %d, got %d",
keyLen, len(op.Key))
}
return nil
}

type eventStoreIter struct {
tableSpan *heartbeatpb.TableSpan
// true when need check whether data from `innerIter` is in `tableSpan`
Expand All @@ -1468,6 +1539,9 @@ type eventStoreIter struct {
decoder *zstd.Decoder
decoderPool *sync.Pool
decodeBuf []byte
// encryptionManager for decrypting data (optional, can be nil).
encryptionManager encryption.EncryptionManager
keyspaceID uint32
}

func (iter *eventStoreIter) Next() (*common.RawKVEntry, bool) {
Expand All @@ -1479,6 +1553,17 @@ func (iter *eventStoreIter) Next() (*common.RawKVEntry, bool) {
key := iter.innerIter.Key()
value := iter.innerIter.Value()

if KeyUsesEncryptionLayer(key) {
if iter.encryptionManager == nil {
log.Panic("encountered encryption-layer value but no encryption manager is configured",
zap.Uint32("keyspaceID", iter.keyspaceID))
}
decryptedValue, err := iter.encryptionManager.DecryptData(context.Background(), iter.keyspaceID, value)
if err != nil {
log.Panic("failed to decrypt value", zap.Error(err))
}
value = decryptedValue
Comment thread
tenfyzhong marked this conversation as resolved.
}
_, compressionType := DecodeKeyAttributes(key)
var decodedValue []byte
if compressionType == CompressionZSTD {
Expand Down
Loading
Loading