Function TestMemoryConnector_ConcurrentCompression

data/memory_compression_test.go:140–244

(t *testing.T)

}

func TestMemoryConnector_ConcurrentCompression(t *testing.T) {
	// Test concurrent access to verify thread safety
	logger := zerolog.New(io.Discard)
	ctx := context.Background()

	connector, err := NewMemoryConnector(ctx, &logger, "test", &common.MemoryConnectorConfig{
		MaxItems: 100_000, MaxTotalSize: "1GB",
	})
	require.NoError(t, err)
	defer connector.Close()

	// Create test data of various sizes
	testData := []struct {
		key   string
		value string
	}{
		{"small1", `{"jsonrpc":"2.0","result":"0x123","id":1}`},
		{"large1", createLargeEvmResponse()},
		{"medium1", createMediumEvmResponse()},
		{"large2", createLargeBlockResponse()},
	}

	// Reduced concurrency to account for Ristretto's async nature
	numWorkers := 10
	numOperationsPerWorker := 5

	var wg sync.WaitGroup
	errChan := make(chan error, numWorkers*numOperationsPerWorker*2) // *2 for Set+Get operations

	// Phase 1: Concurrent writes
	t.Log("Phase 1: Concurrent writes...")
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()

			for j := 0; j < numOperationsPerWorker; j++ {
				for _, testCase := range testData {
					// Unique key per worker and operation
					key := fmt.Sprintf("%s_w%d_op%d", testCase.key, workerID, j)

					// Set operation
					if err := connector.Set(ctx, "concurrent", key, []byte(testCase.value), nil); err != nil {
						errChan <- fmt.Errorf("worker %d operation %d set failed: %w", workerID, j, err)
					}

					// Small delay between operations
					time.Sleep(2 * time.Millisecond)
				}
			}
		}(i)
	}

	// Wait for all writes to complete
	wg.Wait()

	// Allow Ristretto to process all buffered writes
	t.Log("Waiting for Ristretto to process writes...")