Read values and subscription give different values

[lounis-benamar-suez]

Hi, I don't really know if it's a bug. We have a case where we have to use a subscription that could last a very long time. We handle the lost connection issue and other tricks, but I can see that with exactly the same configuration as UAExpert (subscription and monitored item settings) the subscription miss lots of values and sometime nothing change for a day. I tried to use read_values method in parallele of subscription and this return us sometime very old values, I really don't know what is happening. I forgot to mention that we target 900 nodes.

Example of different values between polling and subscription (i.e. nodeId-1234):

time|value|node_id|source
2026-01-27 20:33:39.315072|True|ns=14;nodeId-1234|subscription
2026-01-27 20:33:39.315398|2026-01-27 20:33:39.312882+00:00|i=2258|subscription
2026-01-27 20:33:44.314623|2026-01-27 20:33:44.265685+00:00|i=2258|subscription
2026-01-27 20:33:46.804887|2026-01-27 20:33:46.593481+00:00|i=2258|polling
2026-01-27 20:33:46.805230|False|ns=14;nodeId-1234|polling

Example of blocked subscription (with the opc server time node):

2026-01-26 20:49:41.355702|2026-01-26 20:49:41.305494+00:00|i=2258
2026-01-26 20:49:44.758193|2026-01-26 20:49:44.699704+00:00|i=2258
2026-01-27 08:40:52.209451|2026-01-27 08:40:52.153587+00:00|i=2258
2026-01-27 08:40:54.759477|2026-01-27 08:40:54.675261+00:00|i=2258
2026-01-27 08:41:03.838510|2026-01-27 08:41:03.794363+00:00|i=2258

The code is very simple like this:

class DataChangeHandler:
    def datachange_notification(self, node, val, data):
        try:
            state_queue.put_nowait({
                "node_id": node.nodeid.to_string(),
                "value": val,
                "source_timestamp": data.monitored_item.Value.SourceTimestamp,
                "origin": "subscription",
            })
        except Exception:
            logger.exception("Failed to enqueue subscription value")

    def event_notification(self, event):
        logger.info("Event: %s", event)

    def status_change_notification(self, status):
        logger.info(
            f"Notification status changed. Name: {status.Status.name}, Code: {status.Status.value}, Message: {status.Status.doc}")


async def main_async():
    handler = DataChangeHandler()
    asyncio.create_task(state_worker())
    skip = False
    opcua_client = None

    while True:
        try:
            if skip and opcua_client:
                await opcua_client.client.check_connection()  # ConnectionError('Connection is closed')
                await asyncio.sleep(10)
                continue

            opcua_client = AsyncOpcuaClient(url, username, password)
            await opcua_client.connect()

            if not state or not nodes:
                for node_id in NODE_IDS_TO_SUBSCRIBE:
                    node_instance = opcua_client.client.get_node(node_id)
                    nodes.append(node_instance)
                    state[node_id] = {
                        'instance': node_instance,
                        'source_timestamp': None,
                        'last_value': None,
                    }

            uaParameters = ua.CreateSubscriptionParameters()
            uaParameters.RequestedPublishingInterval = 5000
            uaParameters.RequestedLifetimeCount = 2400
            uaParameters.RequestedMaxKeepAliveCount = 10  # self.get_keepalive_count(period)
            uaParameters.Priority = 255

            await opcua_client.create_subscription(uaParameters, handler)
            await opcua_client.subscribe_data_changes(NODE_IDS_TO_SUBSCRIBE)
            await opcua_client.client.check_connection()  # ConnectionError('Connection is closed')
            skip = True

        except (Exception, ConnectionError) as e:
            logger.exception("Error while monitoring alarms: %s", e)
            skip = False

            try:
                await opcua_client.disconnect()
            except Exception as e:
                logger.error("Unable to disconnect from opcua server: ", e)

            logger.info('Prepare to renew the subscription process to the OPC UA server in 11 seconds.')
            await asyncio.sleep(10)

        finally:
            await asyncio.sleep(1)

The state_worker() is just to write in a file using an async way from asyncio.

MonitoredItem settings:

handles: List[int] = await self.subscription._subscribe(
            nodes=nodes,
            # attr=ua.AttributeIds.Value,
            mfilter=ua.DataChangeFilter(ua.DataChangeTrigger.StatusValueTimestamp),
            queuesize=1,
            # monitoring: ua.MonitoringMode = ua.MonitoringMode.Reporting,
            sampling_interval=2000
        )

In the debug logs I can see a stable state of the program:

INFO - Publish callback called with result...
DEBUG - publish []
DEBUG - Sending: PublishRequest(TypeId...

I noticed that if I re-new the subscription with a different "PublishingInterval" it works again but still bug after a certain time.

[schroeder-]

Your PublishingIntervall is 5000 -> 5seconds. So you will miss changes. You have to lower the PublishingIntervall.
Also you can set sampling_interval=1000 -> 1 Second. This is used to get all values see: sampling_interval
An not forget you can set you queuesize -> https://reference.opcfoundation.org/Core/Part4/v104/docs/5.12.1.5

[lounis-benamar-suez]

Ok, I already tried for example a PublishingInterval with 1000, 500, or 250, queue_size at 100 and sampling_interval at 250, 500, 100. It doesn't explain why the subscription stop receiving any data change ? The connection still running and nothing happened.

For example if a I subscribe on the time node server :

2026-01-26 20:49:41.355702|2026-01-26 20:49:41.305494+00:00|i=2258
2026-01-26 20:49:44.758193|2026-01-26 20:49:44.699704+00:00|i=2258
2026-01-27 08:40:52.209451|2026-01-27 08:40:52.153587+00:00|i=2258
2026-01-27 08:40:54.759477|2026-01-27 08:40:54.675261+00:00|i=2258
2026-01-27 08:41:03.838510|2026-01-27 08:41:03.794363+00:00|i=2258

I didn't know the setting I used for PublishingInterval and SamplingInterval but we can see that there is nothing between 20:49 and 08:40. The logs doesn't show anything useful.

[schroeder-]

Did you set logging to logging.DEBUG and then you look if there are strange log messages in the timeframe where subscriptions are missing.

[lounis-benamar-suez]

I launched a subscription 20 minutes ago,you can see for a specific NodeId:

time|source_time|value|node_id|source
2026-02-03 08:28:38.070356|2026-02-02 15:03:44.668000+00:00|True|the-node-id-1234|subscription

The value the subscription reached is totally outdated (2026-02-02 15:30).

In UAExperts the value is OK in the subscription:

The logs is a reapeatition of this and it seems to be good :

2026-02-03 08:53:38,058 - INFO - Publish callback called with result: PublishResult(SubscriptionId=84, AvailableSequenceNumbers=[], MoreNotifications=False, NotificationMessage_=NotificationMessage(SequenceNumber=2, PublishTime=datetime.datetime(2026, 2, 3, 8, 53, 38, 60505, tzinfo=datetime.timezone.utc), NotificationData=[]), Results=[], DiagnosticInfos=[])
2026-02-03 08:53:38,058 - DEBUG - publish []
2026-02-03 08:53:38,058 - DEBUG - Sending: PublishRequest(TypeId=FourByteNodeId(Identifier=826, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), RequestHeader_=RequestHeader(AuthenticationToken=NodeId(Identifier=b'\x00+z\x04\x80\xc9\x0bj\x97e\xfc\x8ch\x1d8k\xa5\\\x08\x9f)!\x0f\xa9\x15\xeb\x9e)\xe1#\r\xc0', NamespaceIndex=0, NodeIdType=<NodeIdType.ByteString: 5>), Timestamp=datetime.datetime(2026, 2, 3, 8, 53, 38, 58531, tzinfo=datetime.timezone.utc), RequestHandle=1537, ReturnDiagnostics=0, AuditEntryId=None, TimeoutHint=0, AdditionalHeader=ExtensionObject(TypeId=NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), Body=None)), Parameters=PublishParameters(SubscriptionAcknowledgements=[]))
2026-02-03 08:53:38,480 - DEBUG - read
2026-02-03 08:53:38,480 - DEBUG - Sending: ReadRequest(TypeId=FourByteNodeId(Identifier=631, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), RequestHeader_=RequestHeader(AuthenticationToken=NodeId(Identifier=b'\x00+z\x04\x80\xc9\x0bj\x97e\xfc\x8ch\x1d8k\xa5\\\x08\x9f)!\x0f\xa9\x15\xeb\x9e)\xe1#\r\xc0', NamespaceIndex=0, NodeIdType=<NodeIdType.ByteString: 5>), Timestamp=datetime.datetime(2026, 2, 3, 8, 53, 38, 480366, tzinfo=datetime.timezone.utc), RequestHandle=1538, ReturnDiagnostics=0, AuditEntryId=None, TimeoutHint=4000, AdditionalHeader=ExtensionObject(TypeId=NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), Body=None)), Parameters=ReadParameters(MaxAge=0, TimestampsToReturn_=<TimestampsToReturn.Source: 0>, NodesToRead=[ReadValueId(NodeId_=NodeId(Identifier=2259, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), AttributeId=<AttributeIds.Value: 13>, IndexRange=None, DataEncoding=QualifiedName(NamespaceIndex=0, Name=None))]))
2026-02-03 08:53:38,482 - DEBUG - ReadResponse(TypeId=NodeId(Identifier=634, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), ResponseHeader_=ResponseHeader(Timestamp=datetime.datetime(2026, 2, 3, 8, 53, 38, 478509, tzinfo=datetime.timezone.utc), RequestHandle=1538, ServiceResult=StatusCode(value=0), ServiceDiagnostics=DiagnosticInfo(SymbolicId=None, NamespaceURI=None, Locale=None, LocalizedText=None, AdditionalInfo=None, InnerStatusCode=None, InnerDiagnosticInfo=None), StringTable=[], AdditionalHeader=ExtensionObject(TypeId=NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), Body=None)), Results=[DataValue(Value=Variant(Value=0, VariantType=<VariantType.Int32: 6>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2026, 2, 3, 8, 53, 38, 478509, tzinfo=datetime.timezone.utc), ServerTimestamp=None, SourcePicoseconds=None, ServerPicoseconds=None)], DiagnosticInfos=[])
2026-02-03 08:53:39,483 - DEBUG - read
2026-02-03 08:53:39,483 - DEBUG - Sending: ReadRequest(TypeId=FourByteNodeId(Identifier=631, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), RequestHeader_=RequestHeader(AuthenticationToken=NodeId(Identifier=b'\x00+z\x04\x80\xc9\x0bj\x97e\xfc\x8ch\x1d8k\xa5\\\x08\x9f)!\x0f\xa9\x15\xeb\x9e)\xe1#\r\xc0', NamespaceIndex=0, NodeIdType=<NodeIdType.ByteString: 5>), Timestamp=datetime.datetime(2026, 2, 3, 8, 53, 39, 483171, tzinfo=datetime.timezone.utc), RequestHandle=1539, ReturnDiagnostics=0, AuditEntryId=None, TimeoutHint=4000, AdditionalHeader=ExtensionObject(TypeId=NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), Body=None)), Parameters=ReadParameters(MaxAge=0, TimestampsToReturn_=<TimestampsToReturn.Source: 0>, NodesToRead=[ReadValueId(NodeId_=NodeId(Identifier=2259, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), AttributeId=<AttributeIds.Value: 13>, IndexRange=None, DataEncoding=QualifiedName(NamespaceIndex=0, Name=None))]))
2026-02-03 08:53:39,484 - DEBUG - ReadResponse(TypeId=NodeId(Identifier=634, NamespaceIndex=0, NodeIdType=<NodeIdType.FourByte: 1>), ResponseHeader_=ResponseHeader(Timestamp=datetime.datetime(2026, 2, 3, 8, 53, 39, 478144, tzinfo=datetime.timezone.utc), RequestHandle=1539, ServiceResult=StatusCode(value=0), ServiceDiagnostics=DiagnosticInfo(SymbolicId=None, NamespaceURI=None, Locale=None, LocalizedText=None, AdditionalInfo=None, InnerStatusCode=None, InnerDiagnosticInfo=None), StringTable=[], AdditionalHeader=ExtensionObject(TypeId=NodeId(Identifier=0, NamespaceIndex=0, NodeIdType=<NodeIdType.TwoByte: 0>), Body=None)), Results=[DataValue(Value=Variant(Value=0, VariantType=<VariantType.Int32: 6>, Dimensions=None, is_array=False), StatusCode_=StatusCode(value=0), SourceTimestamp=datetime.datetime(2026, 2, 3, 8, 53, 39, 478144, tzinfo=datetime.timezone.utc), ServerTimestamp=None, SourcePicoseconds=None, ServerPicoseconds=None)], DiagnosticInfos=[])
2026-02-03 08:53:40,484 - DEBUG - read