Underestimated the effort to get stream restore working properly in cluster mode.
Some good bug fixes and stability improvments.
Signed-off-by: Derek Collison <derek@nats.io>
I noticed that some consumer go routines were left running at the end
of the test suite.
It turns out that there was a race the way the consumer's qch was closed.
Since it was closed and then set to nil, it is possible that the go
routines that are started and then try to capture o.qch would actually
get qch==nil, wich then when doing a select on that nil channel would
block forever.
So we know pass the qch to the 2 go routines loopAndGatherMsgs() and
loopAndDeliverMsgs() so that when we close the channel there is
no risk of that race happening.
I do believe that there is still something that should be looked at:
it seems that a consumer's delivery loop can now be started/stopped
many times based on leadership acquired/lost. If that is the case,
I think that the consumer should wait for previous go routine to
complete before trying to start new ones.
Also moved 3 JetStream tests to the test/norace_test.go file because
they would consumer several GB of memory when running with the -race flag.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Had a deadlock with new preconditions. We need to hold lock across Store() call but that call could call into storeUpdate() such that we may need to acquire the lock. We can enter this callback from the storage layer itself and the lock would not be held so added an atomic.
Signed-off-by: Derek Collison <derek@nats.io>
- Added non-public stream and consumer configuration options to
achieve the "no subject" and "no interest" capabilities. Had
to implement custom FileStreamInfo and FileConsumerInfo marshal/
unmarshal methods so that those non public fields can be
persisted/recovered properly.
- Restored some of JS original code (since now can use config
instead of passing booleans to the functions).
- Use RLock for deliveryFormsCycle() check (unrelated to MQTT).
- Removed restriction on creating streams with MQTT prefix.
- Preventing API deletion of internal streams and their consumers.
- Added comment on Sublist's ReverseMatch method.
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
This PR introduces native support for MQTT clients. It requires use
of accounts with JetStream enabled. Since as of now clustering is
not available, MQTT will be limited to single instance.
Only QoS 0 and 1 are supported at the moment. MQTT clients can
exchange messages with NATS clients and vice-versa.
Since JetStream is required, accounts with JetStream enabled must
exist in order for an MQTT client to connect to the NATS Server.
The administrator can limit the users that can use MQTT with the
allowed_connection_types option in the user section. For instance:
```
accounts {
mqtt {
users [
{user: all, password: pwd, allowed_connection_types: ["STANDARD", "WEBSOCKET", "MQTT"]}
{user: mqtt_only, password: pwd, allowed_connection_types: "MQTT"}
]
jetstream: enabled
}
}
```
The "mqtt_only" can only be used for MQTT connections, which the user
"all" accepts standard, websocket and MQTT clients.
Here is what a configuration to enable MQTT looks like:
```
mqtt {
# Specify a host and port to listen for websocket connections
#
# listen: "host:port"
# It can also be configured with individual parameters,
# namely host and port.
#
# host: "hostname"
port: 1883
# TLS configuration section
#
# tls {
# cert_file: "/path/to/cert.pem"
# key_file: "/path/to/key.pem"
# ca_file: "/path/to/ca.pem"
#
# # Time allowed for the TLS handshake to complete
# timeout: 2.0
#
# # Takes the user name from the certificate
# #
# # verify_an_map: true
#}
# Authentication override. Here are possible options.
#
# authorization {
# # Simple username/password
# #
# user: "some_user_name"
# password: "some_password"
#
# # Token. The server will check the MQTT's password in the connect
# # protocol against this token.
# #
# # token: "some_token"
#
# # Time allowed for the client to send the MQTT connect protocol
# # after the TCP connection is established.
# #
# timeout: 2.0
#}
# If an MQTT client connects and does not provide a username/password and
# this option is set, the server will use this client (and therefore account).
#
# no_auth_user: "some_user_name"
# This is the time after which the server will redeliver a QoS 1 message
# sent to a subscription that has not acknowledged (PUBACK) the message.
# The default is 30 seconds.
#
# ack_wait: "1m"
# This limits the number of QoS1 messages sent to a session without receiving
# acknowledgement (PUBACK) from that session. MQTT specification defines
# a packet identifier as an unsigned int 16, which means that the maximum
# value is 65535. The default value is 1024.
#
# max_ack_pending: 100
}
```
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
Made several changes based on feedback.
1. Made PubAckResponse only optionally include an ApiError and not force an API type.
2. Allow FilterSubject to be set on a consumer config and cleared if it matches the only stream subject.
3. Remove LookupStream by subject, and add in filters for stream names API.
Signed-off-by: Derek Collison <derek@nats.io>
We would release locks and call into upper layers when removing a message. The upper layers may call back into the lower layers to get more information, such as the subject.
This fix has the storage updates optionally supply the subject for filtered consumers and fixes the bug of double deletes.
Signed-off-by: Derek Collison <derek@nats.io>
In preparation for clustering we need to have the consumer filestore update state with deltas vs original design.
Signed-off-by: Derek Collison <derek@nats.io>
This will track the stream pending state for each consumer.
This code does account for filtered consumers.
Signed-off-by: Derek Collison <derek@nats.io>
- Fix for updating delivery subject and adjusting next delivery sequences.
- When acking explicitly but out of order, need to make sure we set floor correctly.
- Only update ack floors on an ack if the message is present.
- Fix for needAck for explicitAck out of order consumers detecting if message has been acked.
- Fix for race not locking stream when checking interest during stop.
- Fix for filestore determing if a message block still has a message. Added check to first sequence as well as cache.
- Some additions to the original test.
Signed-off-by: Derek Collison <derek@nats.io>
The panic was caused by the closing of an already closed Go channel.
The Delete() relied on the consumer's mset being nil to consider
the consumer already closed. However, the consumer's mset is set
to nil after invoking sendDeleteAdvisoryLocked() which internally
invokes sendAdvisory() which releases/reacquires the consumer lock.
This left an open door for a race to occur and Delete() to be
invoked twice on the same consumer.
Moving setting the consumer's mset to nil too early would prevent
the sendAdvisory() to actually do its job. We could pass the mset
to sendAvisory(), but a simpler approach is to simply use a "closed"
boolean on the Consumer object that is set to true at the beginning
of the Delete() function.
Resolves#1621
Signed-off-by: Ivan Kozlovic <ivan@synadia.com>
This causes nil messages mid content stream and is confusing,
the test for this is amended to test the content received for
validity - where previously the test assumed any content is
good content but in fact it only received half the content.
Removing this behaviour until we can design this properly
Signed-off-by: R.I.Pienaar <rip@devco.net>
