fix(cache): fix cache to HASH - #215
Conversation
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| // ErrGroupAlreadySet groupId already set for PubOptions object | ||
| ErrGroupAlreadySet = errors.New("ErrGroupAlreadySet") | ||
| // ErrInvalidGroupId groupId is invalid | ||
| ErrInvalidGroupId = errors.New("ErrInvalidGroupId") |
There was a problem hiding this comment.
[golint-pr-review] reported by reviewdog 🐶
var ErrInvalidGroupId should be ErrInvalidGroupID
| ) | ||
|
|
||
| type SubResponseHandler = func(msg Message, err error) common.ConsumptionCode | ||
| type SubResponseHandler = func(context context.Context, msg Message, err error) common.ConsumptionCode |
There was a problem hiding this comment.
[golint-pr-review] reported by reviewdog 🐶
exported type SubResponseHandler should have comment or be unexported
|
|
||
| // WithGroup 设置订阅的 GroupId | ||
| // 注意：对于 AtMostOnce 语义，GroupId 在 WithAtMostOnceDelivery 中已设置 | ||
| func WithGroup(groupId common.GroupId) SubOption { |
There was a problem hiding this comment.
[golint-pr-review] reported by reviewdog 🐶
func parameter groupId should be groupID
There was a problem hiding this comment.
Pull Request Overview
This PR addresses caching fixes by switching to a hash‐based implementation and refines related NoSQL document write-back functionality while also polishing message queue subscription options.
- Updated ICache interface and RedisCache implementation to use HGET/HSET for hash storage.
- Introduced a WriteBackWorker for asynchronous cache write-back using message queues.
- Refactored document operations to update or delete cache appropriately and improved subscription option validations.
Reviewed Changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| orm/nosql/worker.go | Added WriteBackWorker for delayed write-back via message queue. |
| orm/nosql/mongo/internal/driver.go | Changed update payload structure by removing extra nesting syntax. |
| orm/nosql/document.go | Updated synchronous and asynchronous save methods and cache updates. |
| orm/nosql/diface/icache.go | Updated ICache interface signatures for hash-based caching. |
| orm/nosql/common.go | Added utility functions for marshaling maps and struct conversions. |
| orm/nosql/cache/redis_cache.go | Refactored GetCache and SetCache to use Redis hash commands. |
| mq/miface/sub_options.go, handler.go, etc. | Adjusted subscription options and handler signatures for consistency. |
There was a problem hiding this comment.
Pull Request Overview
This pull request implements fixes to the cache functionality by switching to a HASH‐based approach while also updating associated document and message queue logic. Key changes include:
- Introducing a WriteBackWorker for asynchronous delayed writeback.
- Modifying the MongoDB driver to update documents using a raw source in the "$set" operation.
- Updating the ICache interface and the Redis cache implementation for HASH-based operations, along with adjustments to MQ handler signatures.
Reviewed Changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| orm/nosql/worker.go | Added WriteBackWorker for asynchronous delayed writeback. |
| orm/nosql/mongo/internal/driver.go | Changed "$set" update to use o.Source directly. |
| orm/nosql/document_test.go | New tests for CRUD and concurrent updates; uses os.Exit(0). |
| orm/nosql/document.go | Adjusted document operations for cache and writeback support. |
| orm/nosql/diface/icache.go | Updated ICache interface to work with HASH-based caching. |
| orm/nosql/cache/redis_cache.go | Revised Redis cache implementation to use HGET/HSET. |
| mq/* (several files) | Updated MQ handler signatures and subscription options. |
Comments suppressed due to low confidence (2)
orm/nosql/document_test.go:126
- Using os.Exit(0) in tests causes premature termination of the test suite. Consider removing these calls to allow all tests to run to completion.
os.Exit(0)
orm/nosql/document_test.go:185
- Using os.Exit(0) in tests causes premature termination of the test suite. Consider removing these calls so that test cleanup and subsequent tests can run.
os.Exit(0)
|
|
||
| update := bson.M{ | ||
| "$set": bson.M{"data": o.Source}, | ||
| "$set": o.Source, |
There was a problem hiding this comment.
The removal of the encapsulating 'data' field in the $set mapping may change the document structure. Ensure that o.Source is already structured as required for the update operation.
| "$set": o.Source, | |
| "$set": bson.M{"data": o.Source}, |
There was a problem hiding this comment.
Pull Request Overview
This PR refactors the caching mechanism to use Redis HASH commands and updates related components, including document caching and message queue handler signatures. Key changes include:
- Updating RedisCache to use HGETALL/HSET for hash-based operations.
- Modifying document caching functions to update or delete cache appropriately.
- Adjusting MQ handler signatures to propagate context.
Reviewed Changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| orm/nosql/worker.go | Introduces WriteBackWorker for delayed database write-back via MQ. |
| orm/nosql/mongo/internal/driver.go | Alters Mongo update document structure by setting o.Source directly. |
| orm/nosql/document.go | Updates cache integration and write-back scheduling logic. |
| orm/nosql/diface/icache.go | Revises ICache interface to work with hash-based cache data formats. |
| orm/nosql/cache/redis_cache.go | Implements Redis cache Get/Set methods using Redis hash commands. |
| mq/miface/handler.go, mq/internal/* | Adds context parameters to MQ handler functions and adjusts related APIs. |
Comments suppressed due to low confidence (3)
orm/nosql/cache/redis_cache.go:64
- The new SetCache method now returns an error, which differs from the previous implementation. Ensure that all callers of SetCache are updated to handle the error return appropriately.
func (c *RedisCache) SetCache(ctx context.Context, key key.Key, data map[string]any, expire time.Duration) error {
orm/nosql/mongo/internal/driver.go:48
- The Mongo update document now directly assigns o.Source to "$set" rather than wrapping it in an object (e.g., {"data": o.Source}). This change could be breaking if the schema expects a nested structure; verify that the document structure aligns with the intended design.
"$set": o.Source,
mq/miface/handler.go:9
- The signature of SubResponseHandler has been updated to include a context parameter. Confirm that all implementations of this handler are adjusted to accept the new signature to avoid runtime issues.
type SubResponseHandler = func(context context.Context, msg Message, err error) common.ConsumptionCode
There was a problem hiding this comment.
Pull Request Overview
This PR fixes the cache implementation to use Redis HASH commands and updates the corresponding document and message queue components accordingly. Key changes include refactoring the cache interface and RedisCache implementation, enhancing the write-back mechanism in DocumentBase and WriteBackWorker, and updating MQ handler signatures to include a context parameter.
- Introduces new write-back logic in orm/nosql/worker.go and updates DocumentBase logic in orm/nosql/document.go.
- Modifies cache interfaces (ICache) and updates the RedisCache implementation to use HSET/HMGET/HGetAll.
- Adjusts MQ handler functions and subscription implementations to pass context parameters.
Reviewed Changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| orm/nosql/worker.go | New write-back worker implementation for delayed document write-back. |
| orm/nosql/mongo/internal/driver.go | Updates the update query to use o.Source directly with "$set" for MongoDB operations. |
| orm/nosql/document_test.go | Adds tests for CRUD and concurrent updates; note the use of os.Exit(0) may affect test execution. |
| orm/nosql/document.go | Enhances caching with updateCache and write-back scheduling functionality. |
| orm/nosql/diface/icache.go | Changes ICache interface methods to return map data and to propagate errors. |
| orm/nosql/cache/redis_cache.go | Refactors RedisCache implementation to use Redis HASH commands for storing cache data. |
| mq/miface/handler.go, mq/internal/* | Updates handler and subscription functions to include context parameters. |
| mq/miface/sub_options.go | Adds additional validation and options adjustments in subscription options. |
| os.Exit(0) | ||
| return nil |
There was a problem hiding this comment.
Using os.Exit(0) in tests can prematurely terminate the test suite and block subsequent tests; consider using t.FailNow() or returning errors instead.
| os.Exit(0) | |
| return nil |
| os.Exit(0) | ||
| return nil |
There was a problem hiding this comment.
The os.Exit(0) call in this test forces early termination; replacing it with proper test assertions will ensure that all tests run.
| os.Exit(0) | |
| return nil |
|
|
||
| update := bson.M{ | ||
| "$set": bson.M{"data": o.Source}, | ||
| "$set": o.Source, |
There was a problem hiding this comment.
Ensure that using o.Source directly in the update query produces the intended schema change; if the original logic expected a nested document (e.g., "data": o.Source), update downstream code accordingly.
| "$set": o.Source, | |
| "$set": bson.M{"data": o.Source}, |
| // 这可能会影响消息的顺序性和处理性能 | ||
| func WithConcurrency(concurrency int) SubOption { | ||
| return func(o *SubOptions) error { | ||
| if concurrency <= 0 { |
There was a problem hiding this comment.
The check for concurrency returns qerrors.ErrInvalidGroupId which is inconsistent with the validation of a concurrency value; consider defining a specific error for invalid concurrency input.
| DefaultCacheTTL = 30 * time.Minute | ||
| // DefaultWriteBackDelay 默认回写延迟时间 | ||
| DefaultWriteBackDelay = 500 * time.Millisecond | ||
| ExpireRangeMin = 6 * time.Hour |
There was a problem hiding this comment.
[golint-pr-review] reported by reviewdog 🐶
exported const ExpireRangeMin should have comment (or a comment on this block) or be unexported
| type WriteBackConfig struct { | ||
| // Enabled 是否启用回写功能 | ||
| Enabled bool `json:"enabled" yaml:"enabled" envconfig:"WRITEBACK_ENABLED" default:"false"` | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| // Delay 回写延迟时间 | ||
| Delay time.Duration `json:"delay" yaml:"delay" envconfig:"WRITEBACK_DELAY" default:"500ms"` | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| // BatchSize 批处理大小 | ||
| BatchSize int `json:"batch_size" yaml:"batch_size" envconfig:"WRITEBACK_BATCH_SIZE" default:"100"` | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| // MaxRetries 最大重试次数 | ||
| MaxRetries int `json:"max_retries" yaml:"max_retries" envconfig:"WRITEBACK_MAX_RETRIES" default:"3"` | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| // RetryDelay 重试延迟 | ||
| RetryDelay time.Duration `json:"retry_delay" yaml:"retry_delay" envconfig:"WRITEBACK_RETRY_DELAY" default:"1s"` | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| mockMQ := &MockMessageQueue{} | ||
| mockProvider := &MockDocumentProvider{} | ||
| logger := zap.NewNop() | ||
|
|
||
| manager, err := NewWriteBackManager(config, mockMQ, mockProvider, logger) | ||
| assert.NoError(t, err) | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
| mockMQ := &MockMessageQueue{} | |
| mockProvider := &MockDocumentProvider{} | |
| logger := zap.NewNop() | |
| manager, err := NewWriteBackManager(config, mockMQ, mockProvider, logger) | |
| assert.NoError(t, err) | |
| mockMQ := &MockMessageQueue{} | |
| mockProvider := &MockDocumentProvider{} | |
| logger := zap.NewNop() | |
| manager, err := NewWriteBackManager(config, mockMQ, mockProvider, logger) | |
| assert.NoError(t, err) | |
|
|
||
| err = manager.Start() | ||
| assert.NoError(t, err) | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| metrics := manager.GetMetrics() | ||
| assert.Equal(t, 0, metrics.WorkerCount) | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
|
|
||
| mockMQ := &MockMessageQueue{} | ||
| mockProvider := &MockDocumentProvider{} | ||
| logger := zap.NewNop() | ||
|
|
||
| manager, err := NewWriteBackManager(config, mockMQ, mockProvider, logger) | ||
| assert.NoError(t, err) | ||
|
|
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
| mockMQ := &MockMessageQueue{} | |
| mockProvider := &MockDocumentProvider{} | |
| logger := zap.NewNop() | |
| manager, err := NewWriteBackManager(config, mockMQ, mockProvider, logger) | |
| assert.NoError(t, err) | |
| mockMQ := &MockMessageQueue{} | |
| mockProvider := &MockDocumentProvider{} | |
| logger := zap.NewNop() | |
| manager, err := NewWriteBackManager(config, mockMQ, mockProvider, logger) | |
| assert.NoError(t, err) | |
| func TestWriteBackPayload_JSON(t *testing.T) { | ||
| payload := WriteBackPayload{ | ||
| CollectionName: "test_collection", | ||
| Key: "test_key", |
There was a problem hiding this comment.
[gofmt] reported by reviewdog 🐶
| Key: "test_key", | |
| Key: "test_key", |
Rebase the document-cache refactor onto current main while keeping CAS, MQ lifecycle, and protobuf-safe document payloads. - Switch ICache/RedisCache to HASH field maps (__version/__data envelope) - Add WriteBack worker/manager with MQ-delayed Mongo persistence - Replace fxmain integration tests with unit tests that do not hang - Reuse mock collections by name so write-back workers see the same store - Update test/orm cache coverage for the new ICache API Co-authored-by: GStones <[email protected]>
6911cdf to
8aad06d
Compare
Replace deprecated reflect.Ptr with reflect.Pointer and simplify the SaveAsync write-back guard for staticcheck QF1001. Co-authored-by: GStones <[email protected]>
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
- Clear legacy Redis string keys on WRONGTYPE during HASH upgrade - Fall back to sync Save when async write-back is disabled - Capture MQ publish deps to avoid DisableWriteBack nil panic - Validate write-back keys without panicking; surface Subscribe errors - Retry out-of-order CAS write-backs; drop only stale snapshots - Round-trip json-tagged fields in map2StructShallow Co-authored-by: GStones <[email protected]>
Rebase early/out-of-order write-backs onto the current DB version instead of nacking forever when an earlier message never lands. On MQ publish failure, SaveAsync now best-effort sync-falls back with the snapshot. Co-authored-by: GStones <[email protected]>
- Use a fresh timeout context for publish-failure sync fallback - Rebase at most one version ahead; reject larger gaps (recreate-safe) - Share applyWriteBackSnapshot between worker and SaveAsync fallback - Avoid double-counting failedCount on rebase errors Co-authored-by: GStones <[email protected]>
Publish the optimistic target version and apply only when it is newer than the DB version. This keeps the latest snapshot under rapid/reordered SaveAsync delivery instead of permanently dropping large version gaps. Co-authored-by: GStones <[email protected]>
Bump a document epoch on Create, store it in the HASH cache, and let workers optionally reject delayed write-backs from a prior generation. Co-authored-by: GStones <[email protected]>
Use unique Create epochs, fence SaveAsync publish fallbacks against the cache generation, and wire cache into WriteBackManager workers. Co-authored-by: GStones <[email protected]>
CAS Set only increments by 1, so gapped targets must be applied repeatedly until DB version matches the optimistic cache version. Also require cache on enabled manager start and reset epoch on init/load. Co-authored-by: GStones <[email protected]>
Avoid DeleteCache during read-through so concurrent Create/SaveAsync generation fences are not wiped; reuse any existing __epoch instead. Co-authored-by: GStones <[email protected]>
Mint a new epoch on cold Load-from-DB, re-check epoch each fast-forward step, abort when an external writer jumps the CAS chain, and make WriteBackManager.Start idempotent. Co-authored-by: GStones <[email protected]>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit c36f318. Configure here.
Check unexpected version jumps before the target comparison so an external writer advancing past the target is not counted as success. Also satisfy staticcheck on the jump test helper. Co-authored-by: GStones <[email protected]>
Note
High Risk
Breaking ICache API plus a new async persistence path that can temporarily diverge cache and DB; epoch/CAS fencing and Redis key-type migration make correctness and rollout sensitive.
Overview
Migrates document caching from JSON strings to Redis HASH fields and adds optional delayed MQ write-back so updates can hit cache first and persist asynchronously.
ICacheis a breaking change:GetCache/SetCachenow operate onmap[string]anyfield maps (with optional field selection) instead of marshaling whole documents.RedisCacheusesHSET/HGETALL/HMGet, clears legacy string keys onWRONGTYPE, and document entries are stored as an envelope (__data,__version,__epoch) with longer jittered TTLs (6–12h).DocumentBasegainsSaveAsync/UpdateAsyncplus epoch fencing so delayed write-backs cannot overwrite a delete/recreate generation. NewWriteBackWorker/WriteBackManagerconsumenats://writeback, fast-forward CAS versions to the optimistic target, drop stale/epoch-mismatched snapshots, and fall back to a sync apply if publish fails.Reviewed by Cursor Bugbot for commit 47900d3. Bugbot is set up for automated code reviews on this repo. Configure here.