Skip to content

fix: catch ValidationError, add stream_add backoff, remove duplicate check #90

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
k-rister merged 3 commits into from
Jun 29, 2026
Merged

fix: catch ValidationError, add stream_add backoff, remove duplicate check #90

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
d505fec
fix: catch ValidationError instead of SchemaError for message validation
k-rister Jun 27, 2026
0b94de9
fix: add backoff on redis connection/timeout errors in stream_add
k-rister Jun 27, 2026
832db54
fix: remove duplicate recipient check in message_for_me
k-rister Jun 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions roadblock.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -549,7 +549,7 @@ def message_validate(self, message):
jsonschema.validate(instance=message, schema=self.schema)
logger.debug("message passed schema validation [%s]", self.message_to_str(message))

except jsonschema.exceptions.SchemaError:
except jsonschema.exceptions.ValidationError:
logger.error("message failed schema validation [%s]", self.message_to_str(message))
return False

Expand All @@ -571,8 +571,6 @@ def message_for_me(self, message):
incomplete_message = True
elif not "id" in message["payload"]["sender"]:
incomplete_message = True
elif not "recipient" in message["payload"]:
incomplete_message = True
elif not "type" in message["payload"]["recipient"]:
incomplete_message = True

Expand Down Expand Up @@ -1237,9 +1235,11 @@ def stream_add(self, stream_name, message):
except redis.exceptions.ConnectionError as con_error:
logger.error("%s", con_error)
logger.error("Stream add to '%s' failed due to connection error!", stream_name)
ret_val = None
except redis.exceptions.TimeoutError as con_error:
logger.error("%s", con_error)
logger.error("Stream add to '%s' failed due to a timeout error!", stream_name)
ret_val = None

if ret_val is None:
logger.warning("Failed attempt %d to add message '%s' to stream '%s'", counter, message, stream_name)
Expand Down Expand Up @@ -1835,7 +1835,7 @@ def run_it(self):

try:
jsonschema.validate(instance=self.user_messages, schema=self.user_schema)
except jsonschema.exceptions.SchemaError as exception:
except jsonschema.exceptions.ValidationError as exception:
logger.critical(exception)
logger.critical("Could not JSON validate the user messages!")
return self.RC_INVALID_INPUT
Expand Down
Loading