Ensure a Message's Response streamId is always set

[frankgh]

patch by Francisco Guerrero; reviewed by TBD for CASSANDRA-21508

[belliottsmith]

I would probably not use a valid streamId as a placeholder you intend to overwrite later

[frankgh]

Yeah, that's a very good point. I will update it

[frankgh]

One thing to consider here is that we won't always overwrite the stream id. I considered using the sentinel value, but this would trip the assertion during the encoding. So we'll need to keep a valid value here.

The cases where the streamId will not be overwritten are:

• ExceptionHandlers.java:81 / PreV5Handlers.java:342 : This occurs during channel negotiations (SSL/TLS handshake failures) and a stream ID is not available
• Unexpected non-wrapped pipeline throwables where no frame/stream ID exists at all

Everything else arrives in a wrapped exception and carries a real stream ID :

• CQLMessageHandler:517 : extracted.streamId() (best-effort)
• CQLMessageHandler:534 (oversized auth) : header.streamId
• CQLMessageHandler:725 (corrupt frame) : NO_REQUEST_STREAM_ID — explicit, the one true context-free CQL path

I think keeping 0 for the NO_REQUEST_STREAM_ID is reasonable because the only sources of missing stream ids are when an error occurs before a stream id is available or when a a corrupt frame does not allow us to extract the stream id.

[belliottsmith]

If we're setting it here, I'm still unsure why we're setting it at all callers to ErrorMessage as well...

[frankgh]

This method is not the only sink for responses, and that's why we still need to set it in the callers to ErrorMessage . For example, the callsites for ExceptionHandlers:81, PreV5Handlers:342, InitialConnectionHandler:134 / 163, and Dispatcher:115 do not flow through processRequest in the Dispatcher, and the stream ID would not be set. I'd prefer to have the redundancy here since it is a harmless operation.

[belliottsmith]

The first two have no streamId. I suspect that InitialConnectionHandler must have a zero streamId as well. I am not sure it is even correct to set it there - it isn't set currently, have you checked the change is valid? That is, it's possible the streamId relates to work to be done after the connection is established, and the error message is meant to be at the connection level? I don't know, but since it is a change in behaviour we should check. Either way, we close the connection immediately in the first case, so we are not at risk of this bug, and we probably should close the connection in the second case as well since it is an unexpected message on startup so the connection is in a corrupted state. I would prefer to simply do this for these two cases anyway, since it's safer without extra validation (no possible change to behaviour, and we won't be at risk of the bug).

That leaves only the Dispatcher case, and you're right that we should address this in a uniform manner. However, this suggests to me that calls to FlushItemConverter should be forced to go through a method that performs this work for us, since this is the final step before writing the reply to the wire, so captures all normal replies. Ideally, we would do something similar for the out-of-band flushes in InitialConnectionHandler and ExceptionHandlers, so that there are at most two routes to send a message. This might be annoying to have the compiler enforce, but I think it would be OK to settle for a uniform invoker of toFlushItem, especially if we are diligent about only setting streamId when we know it, so that we will fail if we hit UNSET_STREAM_ID.

Regarding redundancy: I am not fond of redundancy for its own sake - our goal should always be to make sure the code is well structured so that we must use it correctly, and so we can have confidence that all edge cases are correctly handled (and future edge cases should not be easily added). Redundancy of the kind discussed here implies we aren't confident we can do that, which suggests we need to improve our abstractions instead.

This particular redundancy also has downsides, as it has lead to the use of NO_REQUEST_STREAM_ID in one place where it seems to weaken the protection we are introducing with UNSET_STREAM_ID, because we're enforcing that it be supplied in at least one place where it doesn't seem to make semantic sense.

That said, I won't push hard on this.

[maedhroz]

I think this is too late to set warnings. The finally that resets warnings happens before this. The following test passes again if we move the warning setting back to the pre-patch location:

@Test
    public void warningsAreDeliveredToClient() throws Throwable
    {
        try (Cluster cluster = init(Cluster.build().withNodes(1)
                                           .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL)
                                                                       .set("tombstone_warn_threshold", 0)).start()))
        {
            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int PRIMARY KEY, v int)");
            // insert a row so a tombstone is generated
            cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, v) VALUES (1, 1)", ConsistencyLevel.ONE);
            cluster.coordinator(1).execute("DELETE v FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.ONE);

            String address = cluster.get(1).config().broadcastAddress().getAddress().getHostAddress();
            int port = cluster.get(1).callOnInstance(() -> getNativeTransportPort());

            try (SimpleClient client = SimpleClient.builder(address, port)
                                                   .protocolVersion(ProtocolVersion.V5)
                                                   .build().connect(false))
            {
                QueryMessage query = new QueryMessage("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", QueryOptions.DEFAULT);
                Message.Response response = client.execute(query);

                Assert.assertNotNull("Expected warnings on response, got none", response.getWarnings());
                Assert.assertFalse("Expected non-empty warnings list", response.getWarnings().isEmpty());
            }
        }
    }

[maedhroz]

Actually, I think ClientWarningsTest already catches this.

[frankgh]

this is now fixed after the last commit

[maedhroz]

LGTM, once we fix the bits around warning propagation (see https://github.com/apache/cassandra/pull/4936/changes#r3590923562) and CI is otherwise clean

[belliottsmith]

If we expect this to be overridden, why are we specifying NO_REQUEST_STREAM_ID, rather than UNSET_STREAM_ID?

[frankgh]

There's one case where the negotiation fails during the protocol negotiation (app layer negotiation). This causes the org.apache.cassandra.transport.PreV5Handlers.ExceptionHandler#exceptionCaught code path above. And errorMessage.encode will hit the assertion in org.apache.cassandra.transport.Message#encode. If we set the stream Id to UNSET_STREAM_ID here, this would trigger the assertion error. So we'll need to do special handling for that case. This can be showcased in the org.apache.cassandra.transport.CQLConnectionTest#handleErrorDuringNegotiation test.

[frankgh]

Ok, I spent some time looking at this and I think we can always get a stream id except for one case. I will push a commit with these changes in a bit.

[belliottsmith]

This should not be an assertion, it should be a normal check and throw.

[frankgh]

This has been addressed

[tejalchak]

It seems there is a risk of leaving orphaned trace sessions if there is an unexpected exception from execute. I wonder if moving the traceId to where the decoration/stamping occurs for all cases (toFlushItem or decorateResponse) would make observability exception path get robust as well. Let me know your thoughts (could be fixed in a follow on tkt or here).

[frankgh]

Since this is orthogonal to the problem that we are solving here I think it deserves its own JIRA ticket.

seems there is a risk of leaving orphaned trace sessions if there is an unexpected exception from execute

I don't see how this is possible, the finally block below applies for all exception types. Which means that the tracing will always call stopSession when we are tracing.

I think there might be a case where rewrapping the exception does not propagate the tracing id. So I think it's worth looking into it, but it would be out of scope for this ticket.

[tejalchak]

Wait traceId is very similar to streamId wrt exception handling path being different from normal

try {
    response = execute(queryState, requestTime, shouldTrace);   // ← THROWS here
}
finally {
    if (shouldTrace)
        Tracing.instance.stopSession();          // stops BUT did not 
}
...
if (isTraceable() && isTracingRequested())
    response.setTracingId(tracingSessionId);     // SKIPPED on throw outside try/finally

[frankgh]

Yeah, I think this is an existing issue and it probably deserves its own JIRA. Since it's not a regression in this patch, can we instead fix it in a dedicated ticket?

[tejalchak]

Filed https://issues.apache.org/jira/browse/CASSANDRA-21531

[alexamitin]

NOTE: I know there is a check of streamId for UNSET_STREAM_ID value in Response.encode method of Message.java file.. This check prevents the failure scenario.

However, if somehow the header.streamId==Message.UNSET_STREAM_ID reaches the Envelope.encodeHeaderInto method, the client will receive the response with the stream ID 0.

Here is the reason:

• Message.UNSET_STREAM_ID is defined as Integer.MIN_VALUE (-2147483648).
• Type cast of Integer.MIN_VALUE(-2147483648) to short or byte is 0, which could be a valid stream ID used by the client in the other request.

Suggestion: add the assert to check header.streamdId for the invalid value before adding it to the buffer, which potentially could be send back to the client.

Ditto for Envelope.encodeHeader method.

        assert (header.streamId != Message.UNSET_STREAM_ID) : "Invalid header.streamId";

        if (header.version.isGreaterOrEqualTo(ProtocolVersion.V3))
            buf.putShort((short) header.streamId);
        else
            buf.put((byte) header.streamId);

[frankgh]

Thanks for the suggestion, however every server-side Envelope that reaches the encodeHeader/encodeHeaderInto methods was produced by Message.encode(version, streamId). I do see a potential argument for defense-in-depth, but I think we have some guarantees in place right now that will prevent us from reaching encodeHeaderInto without doing the check first.

[belliottsmith]

Here's a (sketch of a)n alternative approach, where we remove Message.streamId altogether, requiring that the encode call knows it via some deliberate mechanism. It is slightly clunky for SimpleClient, but I think this is a more robust approach, and avoids passing the streamId around to so many places, or redundantly specifying it. The only other difference is that I reverted passing through the startup streamId to the handler, instead opting to pick an arbitrary streamId (0) since we will close the connection anyway; and also I close the connection immediately rather than once the promise completes to ensure this happens immediately after. I am not 100% certain this is necessary, but I would prefer to err on the side of caution.

https://github.com/belliottsmith/cassandra/tree/21508-feedback

[frankgh]

Here's a (sketch of a)n alternative approach, where we remove Message.streamId altogether, requiring that the encode call knows it via some deliberate mechanism. It is slightly clunky for SimpleClient, but I think this is a more robust approach, and avoids passing the streamId around to so many places, or redundantly specifying it. The only other difference is that I reverted passing through the startup streamId to the handler, instead opting to pick an arbitrary streamId (0) since we will close the connection anyway; and also I close the connection immediately rather than once the promise completes to ensure this happens immediately after. I am not 100% certain this is necessary, but I would prefer to err on the side of caution.

https://github.com/belliottsmith/cassandra/tree/21508-feedback

Thanks @belliottsmith , I've applied the commit with a minor follow up. I think we are now in a good spot to merge the patch