Update add queue feature with a more changes and concurrent safety #410

brandur · 2024-06-30T00:20:38Z

A feature to make it possible to add a new queue using an invocation like
client.Queues().Add(name, config), similar to the API for adding a new
periodic job. If the client is started already, the new producer is also
started. Fetch and work context are the same ones created for other
producers during Start.

For now we totally punt on the problem of removing queues, which is more
complicated because it's a fairly hard problem on how producer stop context
cancellation should work.

A problem I was having while writing a stress test for the feature is that
test signal channels were being filled which was difficult to solve because
test signals were always initialized by newTestClient and because the
init also initializes each maintenance service's test signals in turn, it's
not possible to deinit them again. I had to refactor tests such that test
signals are only initialized when they're used, which is probably better
anyway.

brandur · 2024-06-30T00:31:42Z

client.go

+// through Client.Queues.
+type queueBundle struct {
+	// Function that adds a producer to the associated client.
+	addProducer func(queueName string, queueConfig QueueConfig) *producer


Taking a function pointer here might seem a little weird, but it lets us not internalize a client instance, which lets us avoid putting a TTx generic parameter on queueBundle.

brandur · 2024-06-30T00:33:04Z

client_test.go

@@ -155,8 +155,6 @@ func newTestClient(t *testing.T, dbPool *pgxpool.Pool, config *Config) *Client[p
 	client, err := NewClient(riverpgxv5.New(dbPool), config)
 	require.NoError(t, err)

-	client.testSignals.Init()


I was having trouble with my stress test where the test signal channels were filling up, and unfortunately because init'ing test signals also inits test signals for all maintenance services, it means there's no way to deinit them again. The only decent answer is to not init them on every single client initialization, and move inits into test cases that need them.

brandur · 2024-06-30T00:34:17Z

client_monitor.go

@@ -57,6 +57,8 @@ func (m *clientMonitor) Start(ctx context.Context) error {
 // uninitialized.  Unlike SetProducerStatus, it does not broadcast the change
 // and is only meant to be used during initial client startup.
 func (m *clientMonitor) InitializeProducerStatus(queueName string) {
+	m.statusSnapshotMu.Lock()
+	defer m.statusSnapshotMu.Unlock()


Found I needed to add this mutex when trying to run my new stress test. Most other client monitor functions already lock it.

Makes sense. As this wasn't dynamic before it wasn't really needed.

brandur · 2024-06-30T00:38:06Z

@bgentry Mind taking a look at this and letting me know what you think? I definitely found it a little on the tricky side to guarantee we make the right decision to start or not start the new producer based on whether the client is started because it's hard to know for sure whether a client is started or stopped because it may be in the process of stopping or stopping right at this instant (therefore the addition of the new mutex).

I almost think a better design would be to never start a newly added producer/queue, and then require that the caller stop and then start the client again to have it start. However, that would cause some user confusion/surprise I'm sure.

brandur · 2024-06-30T16:41:13Z

@PumpkinSeed BTW, I changed a fair bit of code, but I included all your original commits in this variant.

We're trying a new CLA process (similar to the CLA Assistant you may have seen on any projects) just to make sure we don't have any copyright issues on any of any contributed code. Would you mind helping us test it out by going to https://github.com/riverqueue/cla and following the steps there? It basically involves opening a PR through GitHub's UI and adding your name to a CSV file. Thanks!

brandur · 2024-07-04T04:48:34Z

@bgentry Just since I'm sending 0.9.0 out for a bug fix, I thought I'd ping this one once more. If you have concerns, no worries, but if it's just that you haven't reviewed yet, may be worth including it in the release.

bgentry

LGTM aside from these comments, particularly the Client.Queues() return type which I think is a blocker.

bgentry · 2024-07-06T02:44:14Z

client.go

@@ -1647,6 +1644,10 @@ func (c *Client[TTx]) JobListTx(ctx context.Context, tx TTx, params *JobListPara
 // client, and can be used to add new ones or remove existing ones.
 func (c *Client[TTx]) PeriodicJobs() *PeriodicJobBundle { return c.periodicJobs }

+// Queues returns the currently configured set of queues for the client, and can
+// be used to add new ones.
+func (c *Client[TTx]) Queues() *queueBundle { return c.queues }


I don't think we want to have an exported method that returns an unexported type like this. Whether or not it works, it looks bad/undiscoverable in docs.

I think the type either needs to be exported with minimal methods exposed on it, or else we need to return an exported interface type here.

Ah yeah, makes sense. I'd make it unexported initially because I had to add a TTx generic parameter so it could internalize a client, and it was just a really ugly type. Now that that's gone, totally okay to export this.

bgentry · 2024-07-06T02:45:28Z

client_monitor.go

@@ -57,6 +57,8 @@ func (m *clientMonitor) Start(ctx context.Context) error {
 // uninitialized.  Unlike SetProducerStatus, it does not broadcast the change
 // and is only meant to be used during initial client startup.
 func (m *clientMonitor) InitializeProducerStatus(queueName string) {
+	m.statusSnapshotMu.Lock()
+	defer m.statusSnapshotMu.Unlock()


Makes sense. As this wasn't dynamic before it wasn't really needed.

A feature to make it possible to add a new queue using an invocation like `client.Queues().Add(name, config)`, similar to the API for adding a new periodic job. If the client is started already, the new producer is also started. Fetch and work context are the same ones created for other producers during `Start`. For now we totally punt on the problem of removing queues, which is more complicated because it's a fairly hard problem on how producer stop context cancellation should work. A problem I was having while writing a stress test for the feature is that test signal channels were being filled which was difficult to solve because test signals were always initialized by `newTestClient` and because the init also initializes each maintenance service's test signals in turn, it's not possible to deinit them again. I had to refactor tests such that test signals are only initialized when they're used, which is probably better anyway.

brandur · 2024-07-06T19:34:47Z

@bgentry Updated. Mind taking another look?

brandur · 2024-07-06T20:26:52Z

thx.

brandur force-pushed the brandur-add-queue branch 3 times, most recently from 1fdf021 to d45f42f Compare June 30, 2024 00:30

brandur commented Jun 30, 2024

View reviewed changes

brandur requested a review from bgentry June 30, 2024 00:34

brandur mentioned this pull request Jun 30, 2024

Add AddQueue to the Client #399

Closed

brandur force-pushed the brandur-add-queue branch from d45f42f to 1f11bea Compare June 30, 2024 01:48

brandur force-pushed the brandur-add-queue branch from 1f11bea to e937cb4 Compare July 4, 2024 04:44

brandur force-pushed the brandur-add-queue branch from e937cb4 to 378324a Compare July 4, 2024 14:38

bgentry reviewed Jul 6, 2024

View reviewed changes

PumpkinSeed and others added 4 commits July 6, 2024 12:15

Add AddQueue to the Client

11d77d6

Changes based on reviews

22c507b

Add changes to safely add new Queue

286a9f9

brandur force-pushed the brandur-add-queue branch from 378324a to 89da605 Compare July 6, 2024 19:16

brandur requested a review from bgentry July 6, 2024 19:34

bgentry approved these changes Jul 6, 2024

View reviewed changes

brandur merged commit d4ffd83 into master Jul 6, 2024
10 checks passed

brandur deleted the brandur-add-queue branch July 6, 2024 20:27

PumpkinSeed mentioned this pull request Jul 18, 2024

Dynamic queue handling #396

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Update add queue feature with a more changes and concurrent safety #410

Update add queue feature with a more changes and concurrent safety #410

brandur commented Jun 30, 2024

brandur Jun 30, 2024

brandur Jun 30, 2024

brandur Jun 30, 2024

bgentry Jul 6, 2024

brandur commented Jun 30, 2024

brandur commented Jun 30, 2024

brandur commented Jul 4, 2024

bgentry left a comment

bgentry Jul 6, 2024

brandur Jul 6, 2024

bgentry Jul 6, 2024

brandur commented Jul 6, 2024

brandur commented Jul 6, 2024

Update add queue feature with a more changes and concurrent safety #410

Update add queue feature with a more changes and concurrent safety #410

Conversation

brandur commented Jun 30, 2024

brandur Jun 30, 2024

Choose a reason for hiding this comment

brandur Jun 30, 2024

Choose a reason for hiding this comment

brandur Jun 30, 2024

Choose a reason for hiding this comment

bgentry Jul 6, 2024

Choose a reason for hiding this comment

brandur commented Jun 30, 2024

brandur commented Jun 30, 2024

brandur commented Jul 4, 2024

bgentry left a comment

Choose a reason for hiding this comment

bgentry Jul 6, 2024

Choose a reason for hiding this comment

brandur Jul 6, 2024

Choose a reason for hiding this comment

bgentry Jul 6, 2024

Choose a reason for hiding this comment

brandur commented Jul 6, 2024

brandur commented Jul 6, 2024