Instance having same Provider and Consumer topic name loops causing out of mem

[a-tylenda]

Describe the bug
The Sequence, which is a topic-consumer and topic-provider at the same time, and uses the same topic name for both, like this:

const { PassThrough } = require("stream");

module.exports = [
    {
        requires: "names", 
        contentType: "application/x-ndjson"
    },
    async function(input) {
        const out = new PassThrough({ objectMode: true });

        (await input)
            .map((data) => `Hello ${data.name}!`)
            .pipe(out);

        out.topic = "names";
        out.contentType = "application/x-ndjson";

        return out;
    }
];

After starting such a Sequence and sending some data to topic names (this topic is required by the Sequence) the Instance is caught in an infinite loop, data is being written to topic names and the same data is being read right away. This behavior causes high memory consumption which is not cleared in any way.

Steps to Reproduce

1. run STH locally DEVELOPMENT=1 sth --no-docker

Running STH in docker DEVELOPMENT=1 sth will eventually lead to Runner crush caused by error:

2022-11-09T15:55:50.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 0 ]
2022-11-09T15:55:50.274Z TRACE ServiceDiscovery Topic update. Send topic info to CPM [
  {
    requires: 'names',
    contentType: 'application/x-ndjson',
    topicName: 'names'
  }
]
2022-11-09T15:55:50.274Z DEBUG ServiceDiscovery Get data: [ { topic: 'names', contentType: 'application/x-ndjson' } ]
2022-11-09T15:55:50.274Z DEBUG ServiceDiscovery Topic exists [ 'names' ]
2022-11-09T15:55:50.274Z TRACE ServiceDiscovery Local topic provider not found for: [ 'names' ]
2022-11-09T15:55:51.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 388379 ]
2022-11-09T15:55:52.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 789944 ]
2022-11-09T15:55:53.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 1207697 ]
2022-11-09T15:55:54.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 1648098 ]
2022-11-09T15:55:55.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 2056655 ]
2022-11-09T15:55:56.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 2512294 ]
2022-11-09T15:55:57.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 2933733 ]
2022-11-09T15:55:58.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 3349453 ]
2022-11-09T15:55:59.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 3762779 ]
2022-11-09T15:56:00.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 4135787 ]
2022-11-09T15:56:01.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 4598931 ]
2022-11-09T15:56:02.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 4984669 ]
2022-11-09T15:56:03.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 5410402 ]
2022-11-09T15:56:04.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 5771269 ]
2022-11-09T15:56:05.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 6171219 ]
2022-11-09T15:56:06.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 6622697 ]
2022-11-09T15:56:07.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 7011513 ]
2022-11-09T15:56:08.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 7414978 ]
2022-11-09T15:56:09.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 7805352 ]
2022-11-09T15:56:10.274Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 8211382 ]
2022-11-09T15:56:11.273Z TRACE Auditor Requestor, tx, rx [ 'system', 0, 8647774 ]
2022-11-09T15:57:12.856Z DEBUG SocketServer Channel [75a52a0d-337e-4b55-a4a4-33272658acd4:7] ended 
2022-11-09T15:57:12.857Z DEBUG SocketServer Channel [75a52a0d-337e-4b55-a4a4-33272658acd4:6] ended 
2022-11-09T15:57:12.858Z ERROR SocketServer Error on connection from runner [
  Error: read ECONNRESET
      at TCP.onStreamRead (node:internal/stream_base_commons:217:20) {
    errno: -104,
    code: 'ECONNRESET',
    syscall: 'read'
  }
]
2022-11-09T15:57:12.859Z ERROR SocketServer Error on Instance in stream [
  '75a52a0d-337e-4b55-a4a4-33272658acd4',
  5,
  Error: read ECONNRESET
      at TCP.onStreamRead (node:internal/stream_base_commons:217:20) {
    errno: -104,
    code: 'ECONNRESET',
    syscall: 'read'
  }
]
2022-11-09T15:57:12.860Z DEBUG SocketServer Channel [75a52a0d-337e-4b55-a4a4-33272658acd4:4] ended 
2022-11-09T15:57:12.861Z DEBUG SocketServer Channel [75a52a0d-337e-4b55-a4a4-33272658acd4:3] ended 
2022-11-09T15:57:12.861Z DEBUG SocketServer Channel [75a52a0d-337e-4b55-a4a4-33272658acd4:2] ended 
2022-11-09T15:57:12.861Z DEBUG SocketServer Channel [75a52a0d-337e-4b55-a4a4-33272658acd4:0] ended 
2022-11-09T15:57:13.025Z DEBUG DockerInstanceAdapter Container exited [ 134 ]
2022-11-09T15:57:13.026Z DEBUG DockerInstanceAdapter Container returned non-zero status code [ 134 ]
2022-11-09T15:57:13.026Z ERROR CSIC:75a52a0-... Crashlog [
  [
    '',
    '\n' +
      '<--- Last few GCs --->\n' +
      '\n' +
      '[6:0x5eeb840]   102425 ms: Mark-sweep 252.5 (258.2) -> 251.4 (258.2) MB, 120.9 / 0.0 ms  (average mu = 0.153, current mu = 0.029) allocation failure; scavenge might not succeed\n' +
      '[6:0x5eeb840]   102551 ms: Mark-sweep 252.6 (258.2) -> 251.4 (258.2) MB, 121.8 / 0.0 ms  (average mu = 0.096, current mu = 0.030) allocation failure; scavenge might not succeed\n' +
      '\n' +
      '\n' +
      '<--- JS stacktrace --->\n' +
      '\n' +
      'FATAL ERROR: Ineffective mark-compacts near heap limit Allocation failed - JavaScript heap out of memory\n' +
      ' 1: 0xb6e500 node::Abort() [node]\n' +
      ' 2: 0xa7e632  [node]\n' +
      ' 3: 0xd47ea0 v8::Utils::ReportOOMFailure(v8::internal::Isolate*, char const*, bool) [node]\n' +
      ' 4: 0xd48247 v8::internal::V8::FatalProcessOutOfMemory(v8::internal::Isolate*, char const*, bool) [node]\n' +
      ' 5: 0xf25605  [node]\n' +
      ' 6: 0xf26508 v8::internal::Heap::RecomputeLimits(v8::internal::GarbageCollector) [node]\n' +
      ' 7: 0xf36a13  [node]\n' +
      ' 8: 0xf37888 v8::internal::Heap::CollectGarbage(v8::internal::AllocationSpace, v8::internal::GarbageCollectionReason, v8::GCCallbackFlags) [node]\n' +
      ' 9: 0xf121ee v8::internal::HeapAllocator::AllocateRawWithLightRetrySlowPath(int, v8::internal::AllocationType, v8::internal::AllocationOrigin, v8::internal::AllocationAlignment) [node]\n' +
      '10: 0xf135b7 v8::internal::HeapAllocator::AllocateRawWithRetryOrFailSlowPath(int, v8::internal::AllocationType, v8::internal::AllocationOrigin, v8::internal::AllocationAlignment) [node]\n' +
      '11: 0xef478a v8::internal::Factory::NewFillerObject(int, v8::internal::AllocationAlignment, v8::internal::AllocationType, v8::internal::AllocationOrigin) [node]\n' +
      '12: 0x12b7d2f v8::internal::Runtime_AllocateInYoungGeneration(int, unsigned long*, v8::internal::Isolate*) [node]\n' +
      '13: 0x16e9979  [node]\n'
  ]
]
2022-11-09T15:57:13.026Z INFO  CSIC:75a52a0-... Cleanup completed 
2022-11-09T15:57:13.027Z ERROR CSIC:75a52a0-... Instance promise rejected [ { message: 'Runner failed', exitcode: 134 } ]

2. send seq to sth (topic-consumer-provider-js.tar.gz):

si seq send topic-consumer-provider-js.tar.gz

4. start seq:

si seq start -

5. create file data.json with the content:

{ "name": "Ant-Man" }
{ "name": "Iron Man" }
{ "name": "Hulk" }
{ "name": "Hawkeye" }
{ "name": "Black Widow" }
{ "name": "Thor" }
{ "name": "Captain America" }
{ "name": "Spider-Man" }

6. send this file to topic names:

si topic send names -t application/x-ndjson data.json

7. read topic:

si topic get names

you should get such an output in the console:

(...)
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
"Hello undefined!"
(...)

Expected behavior
The possibility of giving the same name for topic provides and requires should be blocked:

2022-11-09T16:13:09.464Z DEBUG Runner Handshake received 
2022-11-09T16:13:09.470Z TRACE Host PANG received [ { provides: 'names', contentType: 'application/x-ndjson' } ]
2022-11-09T16:13:09.470Z TRACE Host Routing sequence output to topic, name from Sequence [ 'names' ]
2022-11-09T16:13:09.470Z TRACE ServiceDiscovery Routing data: [ { topic: 'names', contentType: '' } ]
2022-11-09T16:13:09.470Z TRACE ServiceDiscovery Topic update. Send topic info to CPM [
  {
    provides: 'names',
    contentType: 'application/x-ndjson',
    topicName: 'names'
  }
]
2022-11-09T16:13:09.466Z DEBUG Runner First Sequence object is not a function: [ { requires: 'names', contentType: 'application/x-ndjson' } ]

Version (please complete the following information):

• STH version:0.30.1
• node version: v16.17.0
• os: Ubuntu 20.4