Js-driver: how to cancel a request

Drivers & Stacks
1

I'd like to provide a filtered readonly view to a large neo4J database. Each time, if the user changes the filter params the previous cypher request will become obsolete and need to be cancelled if it is still running.

Just ignoring the obsolete responses would lead in wasting computing time and is no option.
Cancelling the HTTP request seems to not stop the process in the database and thus is no option as well.

Closing the last open session while the request is still running leads to an exception in the next query. Although I get a new session from the driver I get "you cannot run more transactions on a closed session.". (see example code)

        const delay = (ms: number) => new Promise(resolve => setTimeout(() => resolve(null), ms))
        const time = () => new Date().getTime()
        let t0 = time()
        let log = (msg: string, ...params: unknown[]) => console.log(`${time() - t0}: ${msg}`, ...params)

        let session: Session | null = null

        const userRequests = [
            'match (n) return count(n)',
            'match (a)-[r]->(b) return r',
            'match (m) return m',
            'match (n) return n']

        for (const cypher of userRequests) {
          if (session) {
            log('CLOSE...')
            // await delay(0)
            await session.close()
            log('... CLOSED')
            session = null
          }
          log('RUN...')
          session = d.session()
          session.run(cypher, {})
            .then(res => log(' ... DONE', res.summary.query.text))
            .catch(err => log('... ERROR: ' + (err as Neo4jError).message))
            .finally(() => {
              void session?.close();
              session = null;
            })
        }

console

0: RUN...
0: CLOSE...
4: ... CLOSED
4: RUN...
4: CLOSE...
4: ... ERROR: You cannot run more transactions on a closed session.
6: ... CLOSED
6: RUN...
7: CLOSE...
7: ... ERROR: You cannot run more transactions on a closed session.
9: ... CLOSED
9: RUN...
9: ... ERROR: You cannot run more transactions on a closed session.
11: ... ERROR: You cannot run more transactions on a closed session.

With the 4 requests I would expected 3x ERROR and 1xDONE. But however, maybe closing a session to cancel the request is not intended. Is it?

So, how can I cancel a javascript neo4j driver cypher request?

Many thanks for hints or a solution!

2

You can use await session.close() for closing the session and cancel all the work, this is a way of doing it. Also, the work of a session should not mess with other sessions work.

The problem I'm seeing in the code snippet is the sessions being closed before query run being sent.

This snippet exemplify what is being done:

// you create the session1, this is okay.
let session1: Session? = driver.session()

// fire the query, but this command actually doesn't wait for 
// any command to be dispatched to the server and maybe not even acquired 
// the connection yet
session1.run(cypher)
    .then(res => log(' ... DONE', res.summary.query.text))
    .catch(err => log('... ERROR: ' + (err as Neo4jError).message))
    .finally(async () => {
         // you might need to capture errors in the finally block
         // since errors here can causes issues. 
         await session1?.close();
         session1 = null;
     })

if (session1) {
    // Here is the issue.
    // When you session.close, the session is not open anymore.
    // However, the session1.run didn't fired any command to the server yet.
    // So, when it checks to see if the session is open before send the command,
    // the session will be closed. Then, `session1.run` will reject with the 
    // error reported.
    await session1.close()
}

This kind of error is probably happening because you are simulating a cancellation without wait a bit for cancel (at least let the event loop go to the next iteration).

Rewriting your loop in terms of async iterator/generator. This will simulate the user sending new queries every second, so it will cancel the previous one. The code is in javascript for easy prototyping.


/**
 *. This function emulates a generator with the user sending queries in a infinite stream.
 * 
 * This generator will finish with the last request, but in a real scenario, it will be only interrupted
 * when the user finishes/leave the page/logoff/etc.
 */
async function *getUserRequests() {
   yield 'match (n) return count(n)',
            
   // wait one second
   await new Promise(resolve => setTimeout(resolve, 1000));
  
   yield 'match (a)-[r]->(b) return r';

    // wait one second
   await new Promise(resolve => setTimeout(resolve, 1000));

   yield  'match (m) return m';

   // wait one second
   await new Promise(resolve => setTimeout(resolve, 1000));

   // the user send the last query
   return  'match (n) return n'
}

function createWork (cypher, params) {
  const session = driver.session()
  const context = {
    canceled: false 
  }

  session.run(cypher, params)
        .then(res => { 
          if (context.canceled) {
            log(' ... DONE, but cancelled. Might ignore: ', res.summary.query.text)
            return
          }
          log(' ... DONE', res.summary.query.text)
        })
        .catch(err =>  {
          if (context.canceled) {
            log('... ERROR,  but cancelled. Might ignore: ' + (err as Neo4jError).message)
            return
          }
          log('... ERROR: ' + (err as Neo4jError).message)
        })
        .finally(async () => {
             await session.close();
         })

  return {
    async cancel() {
      context.canceled = true
      await session.close()
    }
  }

}

let lastWork = null

for await  (const cypher of userRequests) {
    if (lastWork) {
      await lastWork.cancel()    
    }
    log('RUN...')
    lastWork = createWork(cypher)
}

Warning! This is draft and this might have bug, the example code is for illustration.

1 Like
3

thanks for your helpful answer!
Your code solves my issue implicitly and is more readable.

My fault was that in the finally function the session pointer points already to the next session instance if the finally is called. Because the session var is in the function context. Storing the session pointer in a var in the for loop context solved the problem.

let cancel: (() => Promise<void>) | null = null
for (const cypher of ['match (a)-[r]-(b) return a', 'match (n) return n']) {
    cancel && await cancel()
    const session = this.driver.session() // solves the issue. Need to store the session pointer in the local context, then the associated final function will get the right session
    cancel = () => session.close()
    session.run(cypher, {})
      .then(res => log('DONE', res.summary.query.text))
      .catch(err => log('ERROR: ' + (err as Neo4jError).message, cypher))
      .finally((async () => await session.close()) as () => void) // why does typscript not accept a Promise as return type as stated in the finally docu
}

I think the exception in your first snippet (my unrolled for loop) is not the issue. Also cancelling in the next event loop iteration does not avoid the exception. However, I think the exception is ok and I just ignore it?

A remaining question:
the typescript type definition for finally is

finally(onfinally?: (() => void) | undefined | null): Promise<T>

so it does not accept an async function like

async () => await session.close()

but the docu says about the finally function:

Its return value is ignored unless the returned value is a rejected promise.

Are the typescript definitions wrong and do I have to work around the type checking by

.finally((async () => await session.close()) as () => void)

improbable, but what is the problem here?

4

Session.close signature is () => Promise<void>.

So, you can do:

.finally(async () => await session.close()) // no need for casting here

The casting to () => void doesn't work because the types are different in this case.

5

yes, I'd expect, what you say as well, but the type definition for finally is

finally(onfinally?: (() => void) | undefined | null): Promise<T>

in

/usr/lib/code/extensions/node_modules/typescript/lib/lib.es2018.promise.d.ts

The type definition seem to not reflect the docu, and I have to work around with the casting to satisfy typescript-eslint
Or I generally misunderstand anything?

However, typescript is probably off topic here?

6

Got it. This seems to be a typescript problem. If you wan't an slighter cleaner workaround, you can omit the error by:

// @ts-expect-error
.finally(async () => await session.close())