Merge pull request #787 from qubic/feature/2026-02-20-oracle-subscriptions

Implement oracle subscriptions

Author: philippwerner

.github/workflows/contract-verify.yml

@@ -35,7 +35,7 @@ jobs:
       - name: Find all contract files to verify
         id: filepaths
         run: |
-          files=$(find src/contracts/ -maxdepth 1 -type f -name "*.h" ! -name "*TestExample*" ! -name "*math_lib*" ! -name "*qpi*" -printf "%p\n" | paste -sd, -)
+          files=$(find src/contracts/ -maxdepth 1 -type f -name "*.h" ! -name "*QUtil*" ! -name "*TestExample*" ! -name "*math_lib*" ! -name "*qpi*" -printf "%p\n" | paste -sd, -)
           echo "contract-filepaths=$files" >> "$GITHUB_OUTPUT"
       - name: Contract verify action step
         id: verify

lib/platform_common/sorting.h

@@ -55,3 +55,176 @@ void quickSort(T* range, int first, int last, SortingOrder order)
 
     return;
 }
+
+template <typename T>
+struct CompareLess
+{
+    constexpr bool operator()(const T& lhs, const T& rhs) const
+    {
+        return lhs < rhs;
+    }
+};
+
+template <typename T>
+struct CompareGreater
+{
+    constexpr bool operator()(const T& lhs, const T& rhs) const
+    {
+        return lhs > rhs;
+    }
+};
+
+// Data structure that provides efficient access to the min element (useful for priority queues).
+template <typename T, unsigned int Capacity, class Compare = CompareLess<T>>
+class MinHeap
+{
+public:
+    void init(Compare compare = Compare())
+    {
+        _size = 0;
+        _compare = compare;
+    }
+
+    // Insert new element. Returns true on success.
+    bool insert(const T& newElement)
+    {
+        if (_size >= capacity())
+            return false;
+        const unsigned int newIndex = _size++;
+        _data[newIndex] = newElement;
+        upHeap(newIndex);
+        return true;
+    }
+
+    // Extract minimum element (copy to output reference and remove it from the heap). Returns true on success.
+    inline bool extract(T& minElement)
+    {
+        return peek(minElement) && drop();
+    }
+
+    // Remove minimum element. Returns true on success.
+    bool drop()
+    {
+        if (_size == 0)
+            return false;
+        _data[0] = _data[--_size];
+        downHeap(0);
+        return true;
+    }
+
+    // Extract minimum element and add a new element in one operation (more efficient than extract followed by insert).
+    // Returns true on success.
+    bool replace(T& minElement, const T& newElement)
+    {
+        if (_size == 0)
+            return false;
+        minElement = _data[0];
+        _data[0] = newElement;
+        downHeap(0);
+        return true;
+    }
+
+    // Remove first elementToRemove that is found. Requires T::operator==(). Least efficient operation of MinHeap with O(N).
+    bool removeFirstMatch(const T& elementToRemove)
+    {
+        // find element
+        unsigned int i = 0;
+        for (; i < _size; ++i)
+        {
+            if (_data[i] == elementToRemove)
+                break;
+        }
+        if (i == _size)
+            return false;
+
+        // replace it by the last element and heapify
+        --_size;
+        if (i < _size)
+        {
+            _data[i] = _data[_size];
+            i = upHeap(i);
+            downHeap(i);
+        }
+        return true;
+    }
+
+    // Get minimum element (copy to output reference and WITHOUT removing it from the heap). Returns true on success.
+    bool peek(T& minElement) const
+    {
+        if (_size == 0)
+            return false;
+        minElement = _data[0];
+        return true;
+    }
+
+    // Return current number of elements in heap.
+    unsigned int size() const
+    {
+        return _size;
+    }
+
+    // Return pointer to elements in heap.
+    const T* data() const
+    {
+        return _data;
+    }
+
+    // Return maximum number of elements that can be stored in the heap.
+    constexpr unsigned int capacity() const
+    {
+        return Capacity;
+    }
+
+protected:
+    void swap(unsigned int idx1, unsigned int idx2)
+    {
+        const T tmp = _data[idx1];
+        _data[idx1] = _data[idx2];
+        _data[idx2] = tmp;
+    }
+
+    unsigned int upHeap(unsigned int idx)
+    {
+        while (idx != 0)
+        {
+            const unsigned int parentIdx = (idx - 1) / 2;
+            if (_compare(_data[idx], _data[parentIdx]))
+            {
+                swap(idx, parentIdx);
+                idx = parentIdx;
+            }
+            else
+            {
+                break;
+            }
+        }
+        return idx;
+    }
+
+    void downHeap(unsigned int idx)
+    {
+        while (1)
+        {
+            const unsigned int childIdx1 = idx * 2 + 1;
+            const unsigned int childIdx2 = childIdx1 + 1;
+            if (childIdx1 >= _size)
+                return;
+            unsigned int minChildIdx = childIdx1;
+            if (childIdx2 < _size && _compare(_data[childIdx2], _data[childIdx1]))
+                minChildIdx = childIdx2;
+            if (_compare(_data[minChildIdx], _data[idx]))
+            {
+                swap(idx, minChildIdx);
+                idx = minChildIdx;
+            }
+            else
+            {
+                return;
+            }
+        }
+    }
+
+    T _data[Capacity];
+    unsigned int _size;
+    Compare _compare;
+};

src/contract_core/contract_def.h

@@ -257,8 +257,6 @@
 // new contracts should be added above this line
 
 #ifdef INCLUDE_CONTRACT_TEST_EXAMPLES
-// forward declaration, defined in qpi_spectrum_impl.h
-static void setContractFeeReserve(unsigned int contractIndex, long long newValue);
 
 constexpr unsigned short TESTEXA_CONTRACT_INDEX = (CONTRACT_INDEX + 1);
 #undef CONTRACT_INDEX
@@ -492,12 +490,6 @@ static void initializeContracts()
     REGISTER_CONTRACT_FUNCTIONS_AND_PROCEDURES(TESTEXB);
     REGISTER_CONTRACT_FUNCTIONS_AND_PROCEDURES(TESTEXC);
     REGISTER_CONTRACT_FUNCTIONS_AND_PROCEDURES(TESTEXD);
-
-    // fill execution fee reserves for test contracts
-    setContractFeeReserve(TESTEXA_CONTRACT_INDEX, 100000000000);
-    setContractFeeReserve(TESTEXB_CONTRACT_INDEX, 100000000000);
-    setContractFeeReserve(TESTEXC_CONTRACT_INDEX, 100000000000);
-    setContractFeeReserve(TESTEXD_CONTRACT_INDEX, 100000000000);
 #endif
 }
 

src/contract_core/qpi_oracle_impl.h

@@ -70,11 +70,12 @@ QPI::sint64 QPI::QpiContextProcedureCall::__qpiQueryOracle(
 	}
 #endif
 
-	// notify about error (status and queryId are 0, indicating that an error happened before sending query)
+	// notify about error (status is 0 and queryId is -1, indicating that an error happened before sending query)
 	auto* state = (ContractStateType*)contractStates[contractIndex];
 	auto* input = (QPI::OracleNotificationInput<OracleInterface>*)__qpiAllocLocals(sizeof(QPI::OracleNotificationInput<OracleInterface>));
 	input->status = ORACLE_QUERY_STATUS_UNKNOWN;
 	input->queryId = -1;
+	input->subscriptionId = -1;
 	QPI::NoData output;
 	auto* locals = (LocalsType*)__qpiAllocLocals(sizeof(LocalsType));
 	notificationProcPtr(*this, *state, *input, output, *locals);
@@ -87,22 +88,114 @@ template <typename OracleInterface, typename ContractStateType, typename LocalsT
 inline QPI::sint32 QPI::QpiContextProcedureCall::__qpiSubscribeOracle(
 	const OracleInterface::OracleQuery& query,
 	void (*notificationProcPtr)(const QPI::QpiContextProcedureCall& qpi, ContractStateType& state, OracleNotificationInput<OracleInterface>& input, NoData& output, LocalsType& locals),
-	QPI::uint32 notificationIntervalInMilliseconds,
 	unsigned int notificationProcId,
+	QPI::uint32 notificationPeriodInMilliseconds,
 	bool notifyWithPreviousReply
 ) const
 {
+	// check that query has timestamp member
 	static_assert(sizeof(query.timestamp) == sizeof(DateAndTime));
-	// TODO
+
+	// check that size of oracle query, oracle reply, notification procedure locals are valid
+	static_assert(sizeof(OracleInterface::OracleQuery) <= MAX_ORACLE_QUERY_SIZE);
+	static_assert(sizeof(OracleInterface::OracleReply) <= MAX_ORACLE_REPLY_SIZE);
+	static_assert(sizeof(LocalsType) <= MAX_SIZE_OF_CONTRACT_LOCALS);
+
+	// check that oracle interface and notification input type are as expected
+	static_assert(sizeof(QPI::OracleNotificationInput<typename OracleInterface::OracleReply>) == 16 + sizeof(OracleInterface::OracleReply));
+	static_assert(OracleInterface::oracleInterfaceIndex < OI::oracleInterfacesCount);
+	static_assert(OI::oracleInterfaces[OracleInterface::oracleInterfaceIndex].replySize == sizeof(OracleInterface::OracleReply));
+	static_assert(OI::oracleInterfaces[OracleInterface::oracleInterfaceIndex].querySize == sizeof(OracleInterface::OracleQuery));
+
+	// check contract index
+	ASSERT(this->_currentContractIndex < 0xffff);
+	const QPI::uint16 contractIndex = static_cast<QPI::uint16>(this->_currentContractIndex);
+
+	// check callback
+	if (!notificationProcPtr || ContractStateType::__contract_index != contractIndex)
+		return -1;
+
+	// check vs registry of user procedures for notification
+	const UserProcedureRegistry::UserProcedureData* procData;
+	if (!userProcedureRegistry || !(procData = userProcedureRegistry->get(notificationProcId)) || procData->procedure != (USER_PROCEDURE)notificationProcPtr)
+		return -1;
+	ASSERT(procData->inputSize == sizeof(OracleNotificationInput<OracleInterface>));
+	ASSERT(procData->localsSize == sizeof(LocalsType));
+
+	// get and destroy fee (not adding to contracts execution fee reserve)
+	const sint64 fee = OracleInterface::getSubscriptionFee(query, notificationPeriodInMilliseconds);
+	const int contractSpectrumIdx = ::spectrumIndex(this->_currentContractId);
+	if (fee >= 0 && contractSpectrumIdx >= 0 && decreaseEnergy(contractSpectrumIdx, fee))
+	{
+		// log burning of QU
+		const QuTransfer quTransfer = { this->_currentContractId, m256i::zero(), fee };
+		logger.logQuTransfer(quTransfer);
+
+		// try to subscribe
+		QPI::sint32 subscriptionId = oracleEngine.startContractSubscription(
+			contractIndex, OracleInterface::oracleInterfaceIndex,
+			&query, sizeof(query), notificationPeriodInMilliseconds,
+			notificationProcId, offsetof(OracleInterface::OracleQuery, timestamp));
+		if (subscriptionId >= 0)
+		{
+			// success
+			if (notifyWithPreviousReply)
+			{
+				// notify contract with previous reply if any is available
+				const OracleSubscription* subscription = oracleEngine.getOracleSubscription(subscriptionId);
+				if (subscription && subscription->lastRevealedQueryId >= 0)
+				{
+					const int64_t queryId = subscription->lastRevealedQueryId;
+					auto* state = (ContractStateType*)contractStates[contractIndex];
+					auto* input = (QPI::OracleNotificationInput<OracleInterface>*)__qpiAllocLocals(sizeof(QPI::OracleNotificationInput<OracleInterface>));
+					input->status = ORACLE_QUERY_STATUS_SUCCESS;
+					input->queryId = queryId;
+					input->subscriptionId = subscriptionId;
+					oracleEngine.getOracleReply(queryId, &input->reply, sizeof(input->reply));
+					QPI::NoData output;
+					auto* locals = (LocalsType*)__qpiAllocLocals(sizeof(LocalsType));
+					notificationProcPtr(*this, *state, *input, output, *locals);
+					__qpiFreeLocals();
+					__qpiFreeLocals();
+				}
+			}
+
+			return subscriptionId;
+		}
+		else if (fee > 0)
+		{
+			// failure -> refund fee
+			oracleEngine.refundFees(_currentContractId, fee);
+		}
+	}
+#if !defined(NDEBUG) && !defined(NO_UEFI)
+	else
+	{
+		addDebugMessage(L"Cannot start contract oracle subscription due to fee issue!");
+	}
+#endif
+
+	// notify about error (status is 0 and subscriptionId is -1, indicating that subscribing failed)
+	auto* state = (ContractStateType*)contractStates[contractIndex];
+	auto* input = (QPI::OracleNotificationInput<OracleInterface>*)__qpiAllocLocals(sizeof(QPI::OracleNotificationInput<OracleInterface>));
+	input->status = ORACLE_QUERY_STATUS_UNKNOWN;
+	input->queryId = -1;
+	input->subscriptionId = -1;
+	QPI::NoData output;
+	auto* locals = (LocalsType*)__qpiAllocLocals(sizeof(LocalsType));
+	notificationProcPtr(*this, *state, *input, output, *locals);
+	__qpiFreeLocals();
+	__qpiFreeLocals();
 	return -1;
 }
 
 inline bool QPI::QpiContextProcedureCall::unsubscribeOracle(
 	QPI::sint32 oracleSubscriptionId
 ) const
 {
-	// TODO
-	return false;
+	ASSERT(this->_currentContractIndex < 0xffff);
+	const QPI::uint16 contractIndex = static_cast<QPI::uint16>(this->_currentContractIndex);
+	return oracleEngine.stopContractSubscription(oracleSubscriptionId, contractIndex);
 }
 
 template <typename OracleInterface>