Controlplane api

[Marco-Premier]

Add WorkItemsService to handle the creation of work items through RabbitMQ queues.
The service provides queue validation, persistent message publishing, and proper
channel recovery after failures.

[wfa-reviewable]

This change is 

[stevenwarejones]

Reviewed 6 of 7 files at r1, all commit messages.
Reviewable status: 6 of 7 files reviewed, 3 unresolved discussions (waiting on @kungfucraig, @Marco-Premier, and @SanjayVas)

---

MODULE.bazel line 236 at r1 (raw file):

        # RabbitMQ
        # TODO(@marcopremier): Remove this once common-jvm PR #286 is merged

can you instead reference your common-jvm commit in this pr temporarily?

---

imports/java/com/rabbitmq/client/BUILD.bazel line 1 at r1 (raw file):

# TODO(@marcopremier): Remove this once common-jvm PR #286 is merged

ditto

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 84 at r1 (raw file):

    } catch (e: Exception) {
      throw StatusException(
        Status.PERMISSION_DENIED.withDescription("Queue '$queueName' does not exist")

is this a NOT_FOUND error? It feels like there could be more reasons than just the queue name not existing that would cause an error.

[kungfucraig]

Please follow the conventional commit guidelines.

Reviewed all commit messages.
Reviewable status: 6 of 7 files reviewed, 9 unresolved discussions (waiting on @Marco-Premier and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 42 at r1 (raw file):

    }

  private lateinit var connection: Connection

Do these need to be lateinit? They are initialized in the init block.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 55 at r1 (raw file):

    } catch (e: Exception) {
      throw StatusException(
        Status.UNAVAILABLE.withDescription("Failed to connect to RabbitMQ: ${e.message}")

Write the host, port, and user name to the message?

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 76 at r1 (raw file):

    val workItem = request.workItem
    val queueName = workItem.queue

Check that this is a valid queue before using it. Or is this what the call to the channel below is doing? Seems like relying on rabbit config for this may not be ideal.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 90 at r1 (raw file):

    try {
      // Makes the message persistent.
      val props = AMQP.BasicProperties.Builder().deliveryMode(2).build()

Please use a constant instead of "2"

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 95 at r1 (raw file):

    } catch (e: Exception) {
      throw StatusException(
        Status.INTERNAL.withDescription("Failed to enqueue work item: ${e.message}")

Write the queue name to the message.

---

src/test/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsServiceTest.kt line 47 at r1 (raw file):

  private lateinit var workItemsService: WorkItemsService

  @Before

Add a testing utility to common JVM to do all of these things. I think this is the third time I've seen this block of code, and would expect to see it many more times. :)

[stevenwarejones]

Reviewed 9 of 14 files at r2, all commit messages.
Reviewable status: 9 of 14 files reviewed, 9 unresolved discussions (waiting on @Marco-Premier and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsService.kt line 31 at r2 (raw file):

class GooglePubSubWorkItemsService(
  private val googlePubSubClient: GooglePubSubClient = DefaultGooglePubSubClient()
//  private val rabbitMqHost: String,

delete these

[Marco-Premier]

Reviewable status: 2 of 14 files reviewed, 8 unresolved discussions (waiting on @kungfucraig, @SanjayVas, and @stevenwarejones)

---

MODULE.bazel line 236 at r1 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

can you instead reference your common-jvm commit in this pr temporarily?

RabbitMQ is no longer used. I removed this line.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsService.kt line 31 at r2 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

delete these

Done.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 42 at r1 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

Do these need to be lateinit? They are initialized in the init block.

Done.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 55 at r1 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

Write the host, port, and user name to the message?

This no longer applies

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 76 at r1 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

Check that this is a valid queue before using it. Or is this what the call to the channel below is doing? Seems like relying on rabbit config for this may not be ideal.

This is done now by the Subscriber in common-jvm. That's it, if a topic doesn't exist an Exception is thrown.

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 84 at r1 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

is this a NOT_FOUND error? It feels like there could be more reasons than just the queue name not existing that would cause an error.

Splitted into more cases

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 90 at r1 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

Please use a constant instead of "2"

no longer applies

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsService.kt line 95 at r1 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

Write the queue name to the message.

no longer applies

---

imports/java/com/rabbitmq/client/BUILD.bazel line 1 at r1 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

ditto

same as above

[stevenwarejones]

Reviewed 6 of 9 files at r3, all commit messages.
Reviewable status: 8 of 14 files reviewed, 9 unresolved discussions (waiting on @kungfucraig, @Marco-Premier, and @SanjayVas)

---

src/main/proto/wfa/measurement/securecomputation/controlplane/v1alpha/work_item.proto line 43 at r3 (raw file):

  // The queue into which this `WorkItem` must be enqueued.
  // Available queues are specified in a configuration file.
  string queue = 2 [

why is this a string now? - do you mean queue_resource_name?

---

src/test/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsServiceTest.kt line 129 at r3 (raw file):

  }

  @Test

can you add a test with multiple queues.... one subscriber per queue?

[stevenwarejones]

Reviewed 2 of 14 files at r2, 2 of 9 files at r3, 2 of 3 files at r4, all commit messages.
Reviewable status: 12 of 14 files reviewed, 8 unresolved discussions (waiting on @kungfucraig, @Marco-Premier, and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsService.kt line 33 at r4 (raw file):

  private val publisher: Publisher = Publisher(projectId, googlePubSubClient)

  override suspend fun createWorkItem(

could this function just be part of the abstract class?

[stevenwarejones]

Reviewable status: 12 of 14 files reviewed, 9 unresolved discussions (waiting on @kungfucraig, @Marco-Premier, and @SanjayVas)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsService.kt line 31 at r4 (raw file):

) : WorkItemsCoroutineImplBase(), WorkItemsService {

  private val publisher: Publisher = Publisher(projectId, googlePubSubClient)

publisher based on the update to common-jvm

[Marco-Premier]

Reviewable status: 8 of 14 files reviewed, 8 unresolved discussions (waiting on @kungfucraig, @SanjayVas, and @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsService.kt line 33 at r4 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

could this function just be part of the abstract class?

Done.

---

src/main/proto/wfa/measurement/securecomputation/controlplane/v1alpha/work_item.proto line 43 at r3 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

why is this a string now? - do you mean queue_resource_name?

I changes it to add new fields to the Queue message but those are actually not needed so I rollback to the initial implementation: #1843

AIP: https://google.aip.dev/122

---

src/test/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/GooglePubSubWorkItemsServiceTest.kt line 129 at r3 (raw file):

Previously, stevenwarejones (Steven Ware Jones) wrote…

can you add a test with multiple queues.... one subscriber per queue?

Done.

[Marco-Premier]

Reviewable status: 7 of 14 files reviewed, 8 unresolved discussions (waiting on @kungfucraig, @SanjayVas, and @stevenwarejones)

---

src/test/kotlin/org/wfanet/measurement/securecomputation/controlplane/v1alpha/WorkItemsServiceTest.kt line 47 at r1 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

Add a testing utility to common JVM to do all of these things. I think this is the third time I've seen this block of code, and would expect to see it many more times. :)

Done.

[kungfucraig]

Reviewed 2 of 14 files at r2, 2 of 9 files at r3, 1 of 3 files at r4, 5 of 5 files at r5, 1 of 1 files at r6, all commit messages.
Reviewable status: 13 of 14 files reviewed, 2 unresolved discussions (waiting on @SanjayVas and @stevenwarejones)