Fix leaking of unqualified viewport ids (#1007)

* Fix leaking of unqualified viewport ids

There were two spots that were leaking unqualified viewport ids.

1. The EditorService was returning the vid when a viewport was unwatched. This is fixed by simply echoing back the oid that the request was submitted with.

2. The Session actor had more significant issues where it was returning an invalid session id, was also returning the internal viewport id, and then was searching for viewport actors with the qualified id.

This should fix both issues and provides a unit test to validate them through the grpc service.

Closes #976

* test cases updated to reproduce the problem reported in #976

---------

Co-authored-by: Davin Shearer <[email protected]>

Author: jw3

packages/client/tests/specs/common.ts

@@ -171,10 +171,13 @@ export async function subscribeViewport(
           : 1
       )
       const event = viewportEvent.getViewportEventKind()
+      const viewport_id_from_event = viewportEvent.getViewportId()
       if (ViewportEventKind.VIEWPORT_EVT_EDIT == event) {
         log_info(
           'viewport_id: ' +
             viewport_id +
+            ', viewport_id_from_event: ' +
+            viewport_id_from_event +
             ', event: ' +
             event +
             ', serial: ' +
@@ -188,6 +191,15 @@ export async function subscribeViewport(
             '", callbacks: ' +
             viewport_callbacks.get(viewport_id)
         )
+        if (viewport_id_from_event !== viewport_id) {
+          log_error(
+            'viewport ID mismatch: "' +
+              viewport_id +
+              '" not equal to "' +
+              viewport_id_from_event +
+              '"'
+          )
+        }
       } else {
         log_info(
           'viewport: ' +

packages/client/tests/specs/viewport.spec.ts

@@ -176,12 +176,15 @@ describe('Viewports', () => {
     await checkCallbackCount(viewport_callbacks, viewport_1_id, 1)
     await checkCallbackCount(viewport_callbacks, viewport_2_id, 3)
     log_info(viewport_callbacks)
+
     // Unsubscribe all viewporta
-    await unsubscribeViewport(viewport_1_id)
-    await unsubscribeViewport(viewport_2_id)
+    expect(await unsubscribeViewport(viewport_1_id)).to.equal(viewport_1_id)
+    expect(await unsubscribeViewport(viewport_2_id)).to.equal(viewport_2_id)
+
     // Note the viewports are not destroyed, just unsubscribed
     expect(await getViewportCount(session_id)).to.equal(1)
-    await destroyViewport(viewport_1_id)
+
+    expect(await destroyViewport(viewport_1_id)).to.equal(viewport_1_id)
     expect(await getViewportCount(session_id)).to.equal(0)
   }).timeout(8000)
 
@@ -304,6 +307,7 @@ describe('Viewports', () => {
       false
     )
     const viewport_id = viewport_response.getViewportId()
+    expect(viewport_id).to.equal(session_id + ':' + desired_viewport_id)
     expect(await viewportHasChanges(viewport_id)).to.be.false
     expect(viewport_response.getData_asU8()).to.deep.equal(Buffer.from(''))
     expect(viewport_response.getLength()).to.equal(0)
@@ -414,7 +418,9 @@ describe('Viewports', () => {
     )
     expect(await getViewportCount(session_id)).to.equal(1)
     // Unsubscribe the viewport
-    expect(await unsubscribeViewport(viewport_id)).to.equal(desired_viewport_id)
+    expect(await unsubscribeViewport(viewport_id)).to.equal(
+      session_id + ':' + desired_viewport_id
+    )
     // Note the viewport is not destroyed, just unsubscribed
     expect(await getViewportCount(session_id)).to.equal(1)
     await destroyViewport(viewport_id)

server/scala/api/src/main/scala/com/ctc/omega_edit/ViewportImpl.scala

@@ -67,6 +67,7 @@ private[omega_edit] class ViewportImpl(p: Pointer, i: FFI) extends Viewport {
 
   def modify(offset: Long, capacity: Long, isFloating: Boolean): Boolean =
     i.omega_viewport_modify(p, offset, capacity, if (isFloating) 1 else 0) == 0
+
   override def toString: String =
     data.mkString // TODO: probably render instead as hex
 

server/scala/serv/src/main/scala/com/ctc/omega_edit/grpc/EditorService.scala

@@ -444,7 +444,7 @@ class EditorService(implicit val system: ActorSystem) extends Editor {
     ObjectId(in.id) match {
       case Viewport.Id(sid, vid) =>
         (editors ? ViewportOp(sid, vid, Viewport.Unwatch)).mapTo[Result].map {
-          case Ok(id) => ObjectId(id)
+          case Ok(_) => in
           case Err(c) => throw grpcFailure(c)
         }
       case _ =>

server/scala/serv/src/main/scala/com/ctc/omega_edit/grpc/Session.scala

@@ -194,7 +194,7 @@ class Session(
       val vid = id.getOrElse(Viewport.Id.uuid())
       val fqid = s"$sessionId:$vid"
 
-      context.child(fqid) match {
+      context.child(vid) match {
         case Some(_) =>
           sender() ! Err(Status.ALREADY_EXISTS)
         case None =>
@@ -204,8 +204,8 @@ class Session(
           val cb = ViewportCallback { (v, e, c) =>
             input.queue.offer(
               ViewportEvent(
-                sessionId = fqid,
-                viewportId = vid,
+                sessionId = sessionId,
+                viewportId = fqid,
                 serial = c.map(_.id),
                 data = Option(ByteString.copyFrom(v.data)),
                 length = Some(v.data.size.toLong),

server/scala/serv/src/test/scala/com/ctc/omega_edit/grpc/Bug976Spec.scala

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2024 Concurrent Technologies Corporation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.ctc.omega_edit.grpc
+
+import com.google.protobuf.ByteString
+import omega_edit.ChangeKind.CHANGE_INSERT
+import omega_edit._
+import org.apache.pekko.stream.Materializer
+import org.apache.pekko.stream.scaladsl.Sink
+import org.scalatest.OptionValues
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AsyncWordSpecLike
+
+import scala.concurrent.duration.DurationInt
+
+class Bug976Spec extends AsyncWordSpecLike with Matchers with OptionValues with EditorServiceSupport {
+  "GitHub issue 976" should useService { implicit svc =>
+    val requested_vid = "my_viewport"
+    implicit val mat = Materializer.matFromSystem(svc.system)
+
+    "validate vid on viewport events and unsubs" in {
+      for {
+        s <- svc.createSession(CreateSessionRequest.defaultInstance)
+        sid = s.sessionId
+        v <- svc.createViewport(CreateViewportRequest(s.sessionId, 1, 0, false, Some(requested_vid)))
+        vid = v.viewportId
+        sub = svc.subscribeToViewportEvents(EventSubscriptionRequest(vid, None))
+        _ = svc.submitChange(ChangeRequest(sid, CHANGE_INSERT, 0, 1, Some(ByteString.fromHex("ff"))))
+        evt <- sub.completionTimeout(1.second).runWith(Sink.headOption)
+        unsub <- svc.unsubscribeToViewportEvents(ObjectId(vid))
+      } yield {
+        vid should startWith(sid)
+        evt.value should matchPattern { case ViewportEvent(`sid`, `vid`, _, _, _, _, _, _) => }
+        unsub.id shouldBe vid
+        val Array(s, v) = vid.split(":")
+        s shouldBe sid
+        v shouldBe requested_vid
+      }
+    }
+  }
+}