feat(pubsub): implement core PubSub framework infrastructure #103
+6,234
−13
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
jbrinkman
force-pushed
the
jbrinkman/pubsub-core
branch
4 times, most recently
from
e6d4562 to
4646a82
Compare
Implements tasks 1-7 of PubSub support specification: Task 1: Core PubSub message types and data structures - Add PubSubMessage class with channel, pattern, and payload support - Add PubSubMessageHandler delegate for callback handling - Add PubSubSubscriptionConfig for subscription configuration Task 2: Message queue implementation - Add PubSubMessageQueue with thread-safe operations - Implement async message retrieval with cancellation support - Add capacity management and overflow handling Task 3: Callback management system - Add PubSubCallbackManager for FFI callback coordination - Implement callback registration and cleanup - Add thread-safe callback invocation Task 4: FFI integration layer - Extend FFI structs with PubSub message and callback types - Add FFI methods for PubSub operations (subscribe, unsubscribe, publish) - Implement callback marshaling and memory management Task 5: Configuration integration - Extend ConnectionConfiguration with PubSub callback support - Add validation for PubSub configuration parameters - Integrate callback setup in connection builder Task 6: BaseClient PubSub foundation - Add PubSub infrastructure to BaseClient - Implement callback manager initialization - Add foundation for PubSub command integration Task 7: Comprehensive unit test coverage - Add tests for all PubSub message types and operations - Add FFI integration and workflow tests - Add configuration and callback management tests - Achieve comprehensive test coverage for core functionality Signed-off-by: Joe Brinkman <[email protected]> # Conflicts: # sources/Valkey.Glide/ConnectionConfiguration.cs
…mory management Add missing Rust FFI implementations for PubSub functionality: - Add PubSubMessageInfo struct for FFI message data - Add PubSubCallback type for callback function pointers - Add register_pubsub_callback() function for callback registration - Add free_pubsub_message() function for memory cleanup - Include placeholder implementations with proper safety documentation These functions provide the Rust-side implementation for the C# FFI declarations added in the previous commit, enabling proper interop between C# PubSub infrastructure and the Rust glide-core library. Signed-off-by: Joe Brinkman <[email protected]>
Replace placeholder PubSub FFI implementation with working callback system: - Add pubsub_callback field to Client struct with thread-safe Mutex protection - Implement proper callback registration in register_pubsub_callback() - Add invoke_pubsub_callback() helper for glide-core integration - Add create_pubsub_message() helper for message creation from Rust strings - Use Arc reference counting for safe client pointer access - Maintain proper memory management and thread safety This provides a complete callback infrastructure ready for integration with glide-core PubSub functionality when it becomes available. Signed-off-by: Joe Brinkman <[email protected]>
- Implement PubSubFFICallbackIntegrationTests with 11 comprehensive test methods - Add ClientRegistry for thread-safe client management with weak references - Test end-to-end message flow from FFI callbacks to message handlers - Test client registry operations under concurrent access with 10 threads - Test callback error handling and recovery with exception isolation - Test memory management and cleanup of marshaled data - Test async message processing without blocking FFI thread pool - Test performance monitoring and logging of callback execution times - Add test collection attributes to prevent parallel execution conflicts - Ensure proper cleanup and disposal of client registry entries Covers requirements 1.1, 1.2, 2.1-2.4, 7.1, 7.4-7.5, 8.1-8.2 from PubSub spec. All 1,781 tests pass successfully with zero failures. Addresses task 7.6: Write integration tests for FFI PubSub callback flow Signed-off-by: Joe Brinkman <[email protected]>
- Simplify test code by removing unnecessary imports - Use collection initializers for lists and arrays - Suppress unnecessary variable assignments with discard operator - Enhance readability of test setup and configuration - Improve error handling and message processing assertions - Remove commented-out code and unnecessary placeholders Refactors existing PubSub FFI callback integration tests to improve code quality, readability, and maintainability while preserving core testing functionality. Signed-off-by: Joe Brinkman <[email protected]>
- Refactor PubSub callback system from static to instance-based approach - Remove `PubSubCallbackManager` and `ClientRegistry` static infrastructure - Update Rust FFI layer to support direct instance callback registration - Modify C# FFI methods and delegates to match new callback signature - Simplify BaseClient PubSub callback handling and lifecycle management - Improve performance by eliminating callback routing and lookup overhead - Align PubSub callback pattern with existing success/failure callback mechanisms - Remove unnecessary client ID tracking and static registration methods Motivation: - Eliminate potential race conditions in callback registration - Reduce code complexity and improve maintainability - Provide a more direct and performant message routing mechanism Signed-off-by: Joe Brinkman <[email protected]>
This commit addresses a critical memory leak in the FFI layer where Rust-allocated memory was not properly freed after C# marshaling during PubSub message processing. Key Changes: Rust FFI Layer (rust/src/lib.rs): - Replaced std::mem::forget() with scoped lifetime management in process_push_notification - Vec<u8> instances now remain alive during callback execution and auto-cleanup on exit - Added comprehensive message structure validation based on PushKind type - Implemented proper error logging for invalid message formats and unexpected value types - Enhanced validation ensures message structure matches expected format for each PushKind Memory Leak Detection Tests: - Added PubSubFFIMemoryLeakTests.cs with comprehensive memory leak detection - Tests process 100,000+ messages and verify memory usage remains bounded - Includes tests for various message sizes, GC pressure, concurrent access, and extended duration - Added PubSubMemoryLeakFixValidationTests.cs for simple validation scenarios Test Cleanup: - Removed tests dependent on deleted PubSubCallbackManager and ClientRegistry classes - Deleted PubSubCallbackIntegrationTests.cs (tested removed PubSubCallbackManager) - Deleted PubSubFFIWorkflowTests.cs (tested removed PubSubCallbackManager) - Deleted ClientRegistryTests.cs (tested removed ClientRegistry) - Updated PubSubFFIIntegrationTests.cs to remove tests using removed infrastructure - Updated PubSubFFICallbackIntegrationTests.cs to remove ClientRegistry tests - Fixed Lock type to object for .NET 8 compatibility - Changed explicit type declarations to var to avoid type resolution issues All unit tests (242) now pass successfully. Addresses requirements 1.1-1.6 and 9.1 from pubsub-critical-fixes spec. Signed-off-by: Joe Brinkman <[email protected]>
- Add volatile modifier to _pubSubHandler field for memory barrier guarantees - Add _pubSubLock object for coordinating thread-safe access - Implement lock-based handler access in HandlePubSubMessage() to prevent race conditions - Update InitializePubSubHandler() with thread-safe initialization - Enhance CleanupPubSubResources() with proper synchronization and timeout-based cleanup - Add thread-safe access to PubSubQueue property Add comprehensive thread safety tests: - Test concurrent message processing from 100+ threads - Test disposal during active message processing - Add stress test with 100 iterations of concurrent operations - Test concurrent access to PubSubQueue and HasPubSubSubscriptions properties - Test rapid create/dispose cycles for memory leak detection - Test disposal timeout handling All 1,771 tests pass (250 unit + 1,521 integration tests). Addresses requirements 2.1-2.6 from pubsub-critical-fixes spec. Signed-off-by: Joe Brinkman <[email protected]>
- Implement bounded channel configuration with PubSubPerformanceConfig - Add configurable capacity (default 1000), backpressure strategies, and shutdown timeout - Replace per-message Task.Run() calls with single dedicated background processing task - Use System.Threading.Channels.Channel<PubSubMessage> for bounded message queuing - Implement non-blocking TryWrite() in PubSubCallback for backpressure handling - Add graceful shutdown with cancellation token support - Create comprehensive performance validation tests covering: - High throughput (10,000+ msg/sec) - Allocation pressure and GC impact - Concurrent message handling - Burst traffic patterns - Long-running stability Benefits: - Single dedicated thread instead of thousands of Task.Run calls - Reduced allocation pressure and GC impact - Predictable performance characteristics - Better resource utilization without thread pool starvation - Configurable performance options for different scenarios All tests pass: 255 unit tests, 1520 integration tests Addresses requirements 3.1-3.6 and 6.1-6.4 from pubsub-critical-fixes spec Signed-off-by: Joe Brinkman <[email protected]>
…nd C# - Add graceful shutdown signaling using tokio::sync::oneshot::channel in Rust - Implement tokio::select! for coordinated task termination in PubSub processing - Store shutdown sender and task handle in Client struct with Mutex for thread safety - Add timeout-based task completion waiting (5 seconds) in close_client - Implement CancellationTokenSource for C# message processing coordination - Add configurable shutdown timeout from PubSubPerformanceConfig - Ensure proper cleanup of channels, tasks, and handlers during disposal - Add comprehensive logging for shutdown process (Debug, Info, Warn levels) - Add unit tests for graceful shutdown coordination - Optimize global usings in test project for cleaner code Validates Requirements: 5.1, 5.2, 5.3, 5.4, 5.5, 5.6, 9.2 Test Results: - All 262 unit tests pass - All 1,772 integration tests pass (1,774 total, 2 skipped) - No regressions introduced Signed-off-by: Joe Brinkman <[email protected]>
…egration tests - Add PubSubQueue property to BaseClient for queue-based message access - Enhance PubSubMessageHandler.GetQueue() to prevent mixing callback/queue modes - Add 11 comprehensive integration tests for queue-based PubSub functionality - Rename PubSubFFICallbackIntegrationTests to PubSubCallbackIntegrationTests - Update unit tests to validate callback/queue mode separation - Add testresults/ and reports/ to .gitignore - Remove obsolete design documentation files Tests cover: - Standalone and cluster client queue retrieval - Pattern subscriptions with queue mode - Message ordering preservation - Async operations with cancellation - High volume message handling - Multiple channel subscriptions - Unicode and special character handling - Error handling for mixed callback/queue modes All 1,797 tests passing with 67.7% line coverage. Signed-off-by: Joe Brinkman <[email protected]>
- Remove unused PubSubConfigurationExtensions class and all extension methods - Remove unused EnableMetrics and MetricsInterval properties from PubSubPerformanceConfig - Simplify Validate() method to only check used properties - Keep core functionality: ChannelCapacity, FullMode, and ShutdownTimeout The extension methods were never integrated into the codebase and the metrics properties were not being used. This cleanup reduces code complexity and maintenance burden while preserving all actively used functionality. Signed-off-by: Joe Brinkman <[email protected]>
- Remove unused using statements from PubSubMessageHandler.cs - Update lock object to use C# 13 Lock type - Apply IDE auto-formatting No functional changes, formatting only. Signed-off-by: Joe Brinkman <[email protected]>
Signed-off-by: Joe Brinkman <[email protected]>
…ge handling - Replace `Lock` with standard `object` for synchronization - Remove unnecessary blank lines in namespace declarations Signed-off-by: Joe Brinkman <[email protected]>
- Populate all PushKind enum values in IsMessageNotification switch - Remove unnecessary using directives (System, System.Collections.Generic) - Remove unused private fields (_callbackExceptions, _receivedMessages) - Fix xUnit2022: Use Assert.False instead of Assert.True with negation - Keep System.Diagnostics using for Stopwatch (not in implicit usings for Lint config) All lint errors resolved. Build passes with --configuration Lint. Signed-off-by: Joe Brinkman <[email protected]>
jbrinkman
force-pushed
the
jbrinkman/pubsub-core
branch
from
ad3669e to
d59edfd
Compare
- Fix Rust FFI logic to create push notification channel when PubSub subscriptions are configured - Previously required both subscriptions AND callback, preventing queue-based message retrieval - Now creates channel whenever subscriptions exist, spawns callback task only when callback provided - Update cluster pattern subscription test to account for cluster mode behavior - Increase subscription establishment wait time from 1s to 5s for cluster propagation - Remove unreliable assertion on PUBLISH return value in cluster mode - Pattern subscriptions may be on different nodes than where channel is hashed This fixes the ClusterPatternSubscription_WithServerPublish_ReceivesMatchingMessages test and enables both callback-based and queue-based PubSub patterns in cluster mode. Signed-off-by: Joe Brinkman <[email protected]>
- Remove SimpleConfigTest.cs (trivial test covered elsewhere) - Remove 2 weak FFI marshaling tests that only verified no exceptions - Remove 1 duplicate Dispose_MultipleCalls_DoesNotThrow test - Remove 2 redundant concurrent processing tests in PubSubThreadSafetyTests - Remove 2 generic timeout tests not specific to PubSub - Consolidate 2 async tests into single comprehensive test - Rename test to accurately reflect behavior (disposal completion vs logging) These changes improve test suite maintainability by removing 8 redundant or misleading tests while maintaining comprehensive coverage (85%+ for PubSub components). All 132 remaining PubSub tests pass successfully. Test Results: - Unit tests: 252 total, 119 PubSub-specific - Integration tests: 1,777 total, 13 PubSub-specific - Combined coverage: 72.7% line, 49.9% branch - PubSub component coverage: 81.8-100% across all classes Signed-off-by: Joe Brinkman <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Overview
This PR implements the core PubSub framework infrastructure for the C# Valkey GLIDE client, providing the foundational components needed to support PubSub functionality.
Addresses: #102
What's Implemented
This PR completes tasks 1-7 of the PubSub framework specification:
✅ Core Infrastructure (Tasks 1-7)
Key Components Added
Message Infrastructure
PubSubMessage: Core message data structure with channel, pattern, and payload supportPubSubMessageHandler: Routes messages to callbacks or queues with comprehensive error handlingPubSubMessageQueue: Thread-safe queue with async retrieval, cancellation, andIAsyncEnumerablesupportConfiguration System
BasePubSubSubscriptionConfig: Base configuration with callback and context supportStandalonePubSubSubscriptionConfig: Standalone client configuration (Exact, Pattern modes)ClusterPubSubSubscriptionConfig: Cluster client configuration (Exact, Pattern, Sharded modes)ConnectionConfigurationviaWithPubSubSubscriptions()FFI Integration
Client Integration
PubSubQueueproperty for queue-based message accessHasPubSubSubscriptionsproperty for subscription statusArchitecture Features
IAsyncEnumerableTesting Coverage
Comprehensive unit tests added covering:
Test Coverage: >95% for all new components
Files Changed
New Files Added (13)
sources/Valkey.Glide/PubSubMessage.cs- Core message data structuresources/Valkey.Glide/PubSubMessageHandler.cs- Message routing handlersources/Valkey.Glide/PubSubMessageQueue.cs- Thread-safe async message queuesources/Valkey.Glide/PubSubSubscriptionConfig.cs- Configuration classessources/Valkey.Glide/Internals/PubSubCallbackManager.cs- FFI callback managementtests/Valkey.Glide.UnitTests/PubSub*.cs(8 test files) - Comprehensive test coverageModified Files (5)
sources/Valkey.Glide/BaseClient.cs- PubSub infrastructure integrationsources/Valkey.Glide/ConnectionConfiguration.cs- PubSub configuration supportsources/Valkey.Glide/Internals/FFI.methods.cs- PubSub FFI methodssources/Valkey.Glide/Internals/FFI.structs.cs- PubSub FFI structurestests/Valkey.Glide.UnitTests/Valkey.Glide.UnitTests.csproj- Test project updatesNext Steps
This framework provides the foundation for:
Breaking Changes
None. This is purely additive functionality that doesn't affect existing APIs.
Checklist
Notes
This PR is in draft mode for initial review. Additional commits may be added based on feedback before marking as ready for review.