Function TestPostgresService_Concurrency

pkg/session/postgres/service_test.go:261–322

TestPostgresService_Concurrency 测试并发安全性

(t *testing.T)

// TestPostgresService_Concurrency 测试并发安全性
func TestPostgresService_Concurrency(t *testing.T) {
	service, cleanup := setupPostgresContainer(t)
	defer cleanup()

	ctx := context.Background()

	sess, err := service.Create(ctx, &session.CreateRequest{
		AppName: "test-app",
		UserID:  "user-001",
		AgentID: "agent-001",
	})
	require.NoError(t, err)

	// 并发追加事件
	const numGoroutines = 10
	const eventsPerGoroutine = 10

	errCh := make(chan error, numGoroutines*eventsPerGoroutine)
	doneCh := make(chan struct{})

	for i := range numGoroutines {
		go func(goroutineID int) {
			for j := range eventsPerGoroutine {
				event := &session.Event{
					ID:           fmt.Sprintf("evt-g%d-e%d", goroutineID, j),
					Timestamp:    time.Now(),
					InvocationID: "inv-concurrent",
					AgentID:      "agent-001",
					Branch:       "root",
					Author:       "user",
					Content: types.Message{
						Role:    types.RoleUser,
						Content: fmt.Sprintf("Message from goroutine %d, event %d", goroutineID, j),
					},
				}

				if err := service.AppendEvent(ctx, sess.ID, event); err != nil {
					errCh <- err
				}
			}
			doneCh <- struct{}{}
		}(i)
	}

	// 等待所有 goroutine 完成
	for range numGoroutines {
		<-doneCh
	}
	close(errCh)

	// 检查错误
	var errors []error
	for err := range errCh {
		errors = append(errors, err)
	}
	assert.Empty(t, errors, "No errors should occur during concurrent operations")

	// 验证所有事件都已插入